随着时间的积累，日志数据会越来越多，当你需要查看并分析庞杂的日志数据时，可通过 Filebeat+Kafka+Logstash+Elasticsearch 采集日志数据到Elasticsearch（简称ES）中，并通过 Kibana 进行可视化展示与分析。
本文介绍具体的实现方法。
Kafka 是一种分布式、高吞吐、可扩展的消息队列服务，广泛用于日志收集、监控数据聚合、流式数据处理、在线和离线分析等大数据领域，已成为大数据生态中不可或缺的部分。在实际应用场景中，为了满足大数据实时检索的需求，一般可以使用 Filebeat 采集日志数据，将 Kafka 作为 Filebeat 的输出端。Kafka 实时接收到 Filebeat 采集的数据后，以 Logstash 作为输出端输出。输出到 Logstash 中的数据在格式或内容上可能不能满足你的需求，此时可以通过 Logstash 的 filter 插件过滤数据。最后将满足需求的数据输出到 ES 中进行分布式检索，并通过 Kibana 进行数据分析与展示。
简单处理流程如下：
CenterOS 7.6 版本，推荐 8G 以上内存。
1、Docker 环境
执行命令如下：
验证下 docker info：
2、Docker Compose 环境
❝Docker Compose是一个用于定义和运行多个 docker 容器应用的工具。使用 Compose 你可以用 YAML 文件来配置你的应用服务，然后使用一个命令，你就可以部署你配置的所有服务了。 ❞
3、版本准备
4、环境初始化
执行命令如下：
5、服务安装
docker-compose.yml 文件内容为：
将该文件上传的 linux 服务器上，执行 docker-compose up 命令即可启动所有服务。
filebeat 客户端安装方式：
6、服务设置
当所有依赖服务启动完成后，需要对以下服务进行一些设置。
修改 filebeat.yml 文件内容
参数说明：
「注意：」客户端 hosts 添加 kafka 对应 server 的 ip 地址 以及 filebeat 配置建议使用 ansible。
「更多配置请参见：」
修改 logstash.conf 内容：
input 参数说明：
output 参数说明：
「注意：」logstash 中最为关键的地方在于 filter,为了调试 filter 的配置。
「更多配置请参见：」
操作命令如下：
字段解释：
从上面的信息可以看出，topic 为 logs 总共消费了 107335 条信息， 未消费的条数为 0。也就是说，消费数据没有积压的情况.
通过 elasticsearch-head 插件查看 ES 中是否收到了由 logstash 发送过来的日志
1、创建 index-pattern
打开 es，进入首页后，点击“connect to your Elasticsearch index”
填入 es 中的索引名，支持正则匹配，输入 Index pattern（本文使用 logs-*），单击 Next step。
选择“@timestamp”作为时间过滤字段，然后点击“create index pattern”：
创建完成后：
2、查看日志
在左侧导航栏，单击 Discover。
从页面左侧的下拉列表中，选择已创建的索引模式（logs-*）。在页面右上角，选择一段时间，查看对应时间段内的 Filebeat 采集的日志数据。
在企业实际项目中，elk 是比较成熟且广泛使用的技术方案。logstash 性能稍弱于 filebeat，一般不直接运行于采集点，推荐使用filebeat。在日志进入elk前，从经验性角度，前置 kafka，一方面作为队列和缓冲，另一方面提供了统一的入口渠道。