如何使用Filebeat+Kafka+Logstash+Elasticsearch构建日志分析系统_检索分析服务Elasticsearch版-阿里云帮助中心

随着时间的积累，日志数据会越来越多，当您需要查看并分析庞杂的日志数据时，可通过Filebeat+Kafka+Logstash+Elasticsearch采集日志数据到阿里云Elasticsearch中，并通过Kibana进行可视化展示与分析。本文介绍具体的实现方法。

背景信息

Kafka是一种分布式、高吞吐、可扩展的消息队列服务，广泛用于日志收集、监控数据聚合、流式数据处理、在线和离线分析等大数据领域，已成为大数据生态中不可或缺的部分。更多信息，请参见什么是云消息队列 Kafka 版。

在实际应用场景中，为了满足大数据实时检索的需求，您可以使用Filebeat采集日志数据，并输出到Kafka中。Kafka实时接收Filebeat采集的数据，并输出到Logstash中。输出到Logstash中的数据在格式或内容上可能不能满足您的需求，此时可以通过Logstash的filter插件过滤数据。最后将满足需求的数据输出到Elasticsearch中进行分布式检索，并通过Kibana进行数据分析与展示。简单流程如下。流程图

操作流程

准备工作

完成环境准备，包括创建阿里云Elasticsearch、Logstash、ECS和
云消息队列 Kafka 版
实例、创建Topic和Consumer Group等。

curl -L -O https://artifacts.elastic.co/downloads/beats/filebeat/filebeat-6.8.5-linux-x86_64.tar.gz
tar xzvf filebeat-6.8.5-linux-x86_64.tar.gz

cd filebeat-6.8.5-linux-x86_64
vi filebeat.kafka.yml

filebeat.prospectors:
  - type: log
    enabled: true
    paths:
        - /var/log/*.log
output.kafka:
    hosts: ["alikafka-post-cn-zvp2n4v7****-1-vpc.alikafka.aliyuncs.com:9092"]
    topic: estest
    version: 0.10.2

参数	说明
type	输入类型。设置为log，表示输入源为日志。
enabled	设置配置是否生效： true：生效 false：不生效
paths	需要监控的日志文件的路径。多个日志可在当前路径下另起一行写入日志文件路径。
hosts	消息队列Kafka实例的单个接入点，可在实例详情页面获取，详情请参见查看接入点。由于本文使用的是VPC实例，因此使用默认接入点中的任意一个接入点。
topic	日志输出到消息队列Kafka的Topic，请指定为您已创建的Topic。
version	Kafka的版本，可在消息队列Kafka的实例详情页面获取。

```
./filebeat -e -c filebeat.kafka.yml
```

input {
  kafka {
    bootstrap_servers => ["alikafka-post-cn-zvp2n4v7****-1-vpc.alikafka.aliyuncs.com:9092,alikafka-post-cn-zvp2n4v7****-2-vpc.alikafka.aliyuncs.com:9092,alikafka-post-cn-zvp2n4v7****-3-vpc.alikafka.aliyuncs.com:9092"]
    group_id => "es-test"
    topics => ["estest"]
    codec => json
filter {
output {
  elasticsearch {
    hosts => "http://es-cn-n6w1o1x0w001c****.elasticsearch.aliyuncs.com:9200"
    user =>"elastic"
    password =>"<your_password>"
    index => "kafka‐%{+YYYY.MM.dd}"
}

表 1. input参数说明
参数	说明
bootstrap_servers	消息队列Kafka实例的接入点，可在实例详情页面获取，详情请参见查看接入点。由于本文使用的是VPC实例，因此使用默认接入点。
group_id	指定为您已创建的Consumer Group的名称。
topics	指定为您已创建的Topic的名称，需要与Filebeat中配置的Topic名称保持一致。
codec	设置为 json ，表示解析JSON格式的字段，便于在Kibana中分析。

表 2. output参数说明
参数	说明
hosts	阿里云Elasticsearch的访问地址，取值为 `http://<阿里云Elasticsearch实例的私网地址>:9200` 。
user	访问阿里云Elasticsearch的用户名，默认为elastic。您也可以使用自建用户，详情请参见通过Elasticsearch X-Pack角色管理实现用户权限管控。
password	访问阿里云Elasticsearch的密码，在创建实例时设置。如果忘记密码，可进行重置，重置密码的注意事项及操作步骤请参见重置实例访问密码。
index	索引名称。设置为 `kafka‐%{+YYYY.MM.dd}` 表示索引名称以kafka为前缀，以日期为后缀，例如 `kafka-2020.05.27` 。

input {
 kafka {
  bootstrap_servers => ["alikafka-post-cn-zvp2n4v7****-1-vpc.alikafka.aliyuncs.com:9092,alikafka-post-cn-zvp2n4v7****-2-vpc.alikafka.aliyuncs.com:9092,alikafka-post-cn-zvp2n4v7****-3-vpc.alikafka.aliyuncs.com:9092"]
  group_id => "es-test"
  topics => ["estest"]
  codec => json
kafka {
  bootstrap_servers => ["alikafka-post-cn-zvp2n4v7****-1-vpc.alikafka.aliyuncs.com:9092,alikafka-post-cn-zvp2n4v7****-2-vpc.alikafka.aliyuncs.com:9092,alikafka-post-cn-zvp2n4v7****-3-vpc.alikafka.aliyuncs.com:9092"]
  group_id => "es-test-2"
  topics => ["estest_2"]
  codec => json
}

原因	解决方法
Kafka的接入点不正确。	参见查看接入点获取正确的接入点。完成后，修改管道配置替换错误接入点。
Logstash与Kafka不在同一VPC下。	重新购买同一VPC下的实例。购买后，修改现有管道配置。
Kafka或Logstash集群的配置太低，例如使用了测试版集群。	升级集群规格，完成后，刷新实例，观察变更进度。升级Logstash实例规格的具体操作，请参见升配集群；升级Kafka实例规格的具体操作，请参见升级实例配置。
管道配置中包含了file_extend，但没有安裝logstash-output-file_extend插件。	选择以下任意一种方式处理：安装logstash-output-file_extend插件。具体操作，请参见安装或卸载插件。中断变更，等到实例处于变更中断状态后，在管道配置中，去掉file_extend配置，触发重启恢复。

参数	说明
管道工作线程	并行执行管道的Filter和Output的工作线程数量。当事件出现积压或CPU未饱和时，请考虑增大线程数，更好地使用CPU处理能力。默认值：实例的CPU核数。
管道批大小	单个工作线程在尝试执行Filter和Output前，可以从Input收集的最大事件数目。较大的管道批大小可能会带来较大的内存开销。您可以设置LS_HEAP_SIZE变量，来增大JVM堆大小，从而有效使用该值。默认值：125。
管道批延迟	创建管道事件批时，将过小的批分派给管道工作线程之前，要等候每个事件的时长，单位为毫秒。默认值：50ms。
队列类型	用于事件缓冲的内部排队模型。可选值： MEMORY ：默认值。基于内存的传统队列。 PERSISTED ：基于磁盘的ACKed队列（持久队列）。
队列最大字节数	请确保该值小于您的磁盘总容量。默认值：1024 MB。
队列检查点写入数	启用持久性队列时，在强制执行检查点之前已写入事件的最大数目。设置为0，表示无限制。默认值：1024。

背景信息

操作流程

准备工作

步骤一：安装并配置Filebeat

步骤二：配置Logstash管道

步骤三：查看日志消费状态

步骤四：通过Kibana过滤日志数据

常见问题