elk日志分析系统kafka（elk+kafka+filebeat）

架构图解架构解读第一层：数据采集层第二层：数据缓冲层第三层：数据转发层第四层：数据持久化存储层第五层：数据检索、展示层部署方案角色设计搭建部署一、安装JDK 1.8二、调整系统参数三、elasticsearch集群1.部署ES yum源2.配置验证KEY：3.安装ES server4.安装head、bigdesk、kopf插件1.安装head插件2.安装bigdesk插件3.安装kopf插件四、Zookeeper集群1.安装Zookeeper2.创建myid3.启动服务 & 查看状态五、Kafka集群1.安装Kfaka2.修改配置文件3.配置对应IP解析4.启动kafka六、Logstash1.安装logstash2.配置logstash3.启动logstash4.测试logstash正常运行4.配置logstash收集nginx日志七、Filebeat1.安装filebeat2.配置filbeat

elk+kafka+filebeat

为什么要做日志分析平台，随着业务量的增大，单个日志文件大小达到几个GB，这时候，我们发现用系统自带的工具，cat/grep/awk/tail越来越显得力不从心，除了操作系统日志，还有应用系统日志，分布在各个服务器上，分析工作异常繁琐。所以为了解决现状，接下来我们将构建这个日志分析平台，具体内容如上：

架构解读

第一层：数据采集层

LOGS层:业务应用服务器集群，使用filebeat对应用服务器的日志进行采集。

第二层：数据缓冲层

zookeeper+kafka层:日志采集客户端采集来的数据，转存到kafka+zookeeper集群中,做一个消息队列，让数据有一定的缓冲

第三层：数据转发层

logstash层:这个单独的Logstash节点会实时去kafka broker集群拉数据，转发至ES DataNode。

第四层：数据持久化存储层

ES层：会把收到的数据，写磁盘，建索引库。

第五层：数据检索、展示层

Kibana层： 主要协调ES集群，处理数据检索请求，数据展示。

部署方案

角色设计

部署方案

角色设计

IP

角色

所属集群

10.10.10.151

应用服务器+filebeat

业务系统

10.10.10.121

Logstash+Kafka+ZooKeeper

kafka broker集群

10.10.10.123

Logstash+Kafka+ZooKeeper

kafka broker集群

10.10.10.125

Logstash+Kafka+ZooKeeper

kafka broker集群

10.10.10.202

elasticsearch+logstash

es 集群

10.10.10.200

elasticsearch+kibana

es 集群

搭建部署

为了节约服务器资源，所以把一下角色合并到一台服务器中去。

系统环境：Centos7+ JDK

一、安装JDK 1.8

1. yum install java
2. # 这里直接采用yum安装，为了节省时间

二、调整系统参数

1. # 配置系统最大打开文件描述符数
2. vim /etc/sysctl.conf
3. fs.file-max=65535
5. # 配置进程最大打开文件描述符
6. vim /etc/security/limits.conf
7. # End of file
8. * soft nofile 65535
9. * hard nofile 65535

10. vim /etc/sysconfig/elasticsearch

11. ES_HEAP_SIZE=16g

12.# 机器的可用内存为32G

三、elasticsearch集群

1.部署ES yum源

这里采用rpm安装方式，安装的方式看个人喜好去选择，这里就不过多强调了，根据官网的安装方式进行安装。需要编辑/etc/yum.repos.d/elasticsearch.repo内容如下：

[elasticsearch-5.x]name=Elasticsearch repository for 5.x packagesbaseurl=https://artifacts.elastic.co/packages/5.x/yumgpgcheck=1gpgkey=https://artifacts.elastic.co/GPG-KEY-elasticsearchenabled=1autorefresh=1type=rpm-md

2.配置验证KEY：

1. rpm --import https://artifacts.elastic.co/GPG-KEY-elasticsearc

3.安装ES server

1. yum install -y elasticsearch

配置Es 配置文件，RPM安装的配置文件默认路径为 /etc/elasticsearch/elasticsearch.yml

1. # 配置ES配置文件
2. vim /etc/elasticsearch/elasticsearch.yml
4. # 内容如下：
5. grep -Ev '(^$|#)' /etc/elasticsearch/elasticsearch.yml
7. cluster.name: YL-Cluster # 集群名称
8. node.name: es-node01 # 节点ID
9. path.data: /data/es/data # ES数据路径
10. path.logs: /data/es/logs # ES日志路径
11. network.host: 10.10.10.202 # 主机地址，可以是IP，也可以主机名，你开心就好
12. http.port: 9200 # 服务监听端口，保持默认即可
13. discovery.zen.ping.unicast.hosts: ["10.10.10.202", "10.10.10.200"]
14. discovery.zen.minimum_master_nodes: 1
15. bootstrap.system_call_filter: false
16. http.cors.enabled: true
17. http.cors.allow-origin: "*"

另外一台ES的配置文件和上述相同，要注意的是修改一下：

 注: path.data、path.logs 这两个参数指定的路径，如果没有需要自己创建，还要赋予权限给elasticsearch用户。ES nodes都是一样的。

1. node.name: es-node02
2. network.host: 10.10.10.200

启动方式这里说一下，采用RPM安装，会自动在/etc/init.d/目录下生成elasticsearch的启动文件所以启动方式如下：

1. /etc/init.d/elasticsearch start
2. 或
3. service elasticsearch start
5. # Centos 7使用一下启动方式
6. systemctl start elasticsearch

4.安装head、bigdesk、kopf插件

由于elasticsearch 5.X不再建议支持插件的安装方式，所以推荐使用独立的方式安装head、bigdesk插件。

1.安装head插件

插件链接：https://github.com/mobz/elasticsearch-head

1. # Install && Running
2. git clone git://github.com/mobz/elasticsearch-head.git
3. cd elasticsearch-head
4. npm install
5. nohub npm run server &
7. # 注意给elasticsearch配置一下两个参数，方便head插件可以访问es
8. http.cors.enabled: true
9. http.cors.allow-origin: "*"

OK,这里安装结束了，打开你的浏览器访问http://headserverIP:9100,效果如下：

2.安装bigdesk插件

插件链接：https://github.com/hlstudio/bigdesk

1. # 大致安装步骤如下：
2. git clone https://github.com/hlstudio/bigdesk
3. cd bigdesk/_site/
4. # 启动服务
5. python -m SimpleHTTPServer
6. 或
7. nohub python -m SimpleHTTPServer &

bigdesk的默认地址为:http://bigdeskserverip:8000,效果如下：

3.安装kopf插件

四、Zookeeper集群

配置10.10.10.121/10.10.10.123/10.10.10.125的zookeeper集群

1.安装Zookeeper

zookeeper官网:  http://zookeeper.apache.org/这里要注意，请下载XXX版，无需安装解压可用，千万不要下载错了。

1. # zookeeper依赖于java，前面已经安装了，这里就不在强调了
2. wget
3. # 解压到/usr/local目录下
4. tar -zxvf zookeeper-3.4.9.tar.gz -C /usr/local
5. # 编写配置文件
6. vim /usr/local/zookeeper-3.4.9/conf/zoo.cfg
8. grep -Ev '(^$|#)' /usr/local/zookeeper-3.4.9/conf/zoo.cfg
10. tickTime=2000
11. initLimit=10
12. syncLimit=5
13. dataDir=/usr/local/zookeeper/data
14. dataLogDir=/usr/local/zookeeper/logs
15. clientPort=2181
16. server.1=zk001:2888:2777
17. server.2=zk002:2888:2777
18. server.3=zk003:2888:2777

同步配置文件到其他两台节点，

zookeeper集群，每个节点的配置都是一样的，不需要做任何更改，不熟悉的zookeeper的小伙伴，可以参考：scp zoo.cfg 10.10.10.123:/usr/local/zookeeper-3.4.9/conf/zoo.cfgscp zoo.cfg 10.10.10.125:/usr/local/zookeeper-3.4.9/conf/zoo.cfg

2.创建myid

1. # 10.10.10.121
2. echo 1 >/usr/local/zookeeper-3.4.9/data/myid
4. # 10.10.10.123
5. echo 2 >/usr/local/zookeeper-3.4.9/data/myid
7. # 10.10.10.125
8. echo 3 >/usr/local/zookeeper-3.4.9/data/myid

3.启动服务 & 查看状态

1. # 10.10.10.121
2. bin/zkServer.sh start
3. bin/zkServer.sh status
5. ZooKeeper JMX enabled by default
6. Using config: /usr/local/zookeeper/zookeeper-3.4.9/bin/../conf/zoo.cfg
7. Mode: leader
9. # 10.10.10.123
10. bin/zkServer.sh start
11. bin/zkServer.sh status
13. ZooKeeper JMX enabled by default
14. Using config: /usr/local/zookeeper/zookeeper-3.4.9/bin/../conf/zoo.cfg
15. Mode: follower
17. # 10.10.10.125
18. bin/zkServer.sh start
19. bin/zkServer.sh status
21. ZooKeeper JMX enabled by default
22. Using config: /usr/local/zookeeper/zookeeper-3.4.9/bin/../conf/zoo.cfg
23. Mode: follower

五、Kafka集群

配置kafka broker集群Kafka官网:  http://kafka.apache.org/

1.安装Kfaka

1. # 下载免安装版本，Binary 版本，记得自己看清楚
2. wget http://219.238.4.227/files/A180000005766BA1/mirrors.hust.edu.cn/apache/kafka/0.10.2.1/kafka_2.12-0.10.2.1.tgz
3. tar -zxvf kafka_2.12-0.10.2.1.tgz -C /usr/local

2.修改配置文件

1. # 10.10.10.121 节点
2. [root@zk001 config]# grep -Ev '(^$|#)' server.properties
4. broker.id=1
5. delete.topic.enable=true
6. listeners=PLAINTEXT://10.10.10.121:9092
7. num.network.threads=8
8. num.io.threads=8
9. socket.send.buffer.bytes=102400
10. socket.receive.buffer.bytes=102400
11. socket.request.max.bytes=104857600
12. log.dirs=/usr/local/kafka/data/kafka-logs
13. num.partitions=20
14. num.recovery.threads.per.data.dir=1
15. log.retention.hours=72
16. log.segment.bytes=1073741824
17. log.retention.check.interval.ms=300000
18. zookeeper.connect=10.10.10.121:2181,10.10.10.123:2181,10.10.10.125:2181
19. zookeeper.connection.timeout.ms=6000

同步配置文件到10.10.10.123/10.10.10.125，内容基本相同，只需要修改一下broker.id和listeners

1. # 同步配置文件
2. scp server.properties 10.10.10.123:/usr/local/kafka/kafka_2.11-0.10.0.1/config/
3. scp server.properties 10.10.10.125:/usr/local/kafka/kafka_2.11-0.10.0.1/config/
5. # 修改broker.id和listeners
6. # 10.10.10.123
7. broker.id=2
8. listeners=PLAINTEXT://10.10.10.123:9092
10. # 10.10.10.125
11. broker.id=3
12. listeners=PLAINTEXT://10.10.10.125:9092

3.配置对应IP解析

1. # 10.10.10.121
2. vim /etc/hosts
3. 10.10.10.121 zk001
4. 10.10.10.123 zk002
5. 10.10.10.125 zk003
7. # 其他两台的的hosts配置也是一样的，记得同步

4.启动kafka

1. cd /usr/local/kafka/bin
2. ./kafka-server-start.sh -daemon /usr/local/kafka/config/server.properties
4. # 其他两个节点的服务启动方式是一样的

到此，kakfa+zookeeper集群搭建完成。

友情赠送：

1. bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test # 创建topic
3. bin/kafka-topics.sh --list --zookeeper localhost:2181 # 查看已经创建的topic列表
5. bin/kafka-topics.sh --describe --zookeeper localhost:2181 --topic test # 查看topic的详细信息
7. bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test # 发送消息, 回车后模拟输入一下消息
9. bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test # 消费消息, 可以换到其他kafka节点, 同步接收生产节点发送的消息
11. bin/kafka-topics.sh --zookeeper localhost:2181 --alter --topic test --partitions 6 # 给topic增加分区
13. bin/kafka-topics.sh --delete --zookeeper localhost:2181 --topic test1 # 删除已经创建的topic, 前提是开了delete.topic.enable=true参数
15. 如果还不能删除, 可以到zookeeper中去干掉它
17. cd /usr/local/zookeeper-3.4.9/
19. bin/zkCli.sh
21. ls /brokers/topics # 查看topic
23. rmr /brokers/topics/test1 # 删除topic

六、Logstash

配置10.10.10.121 节点

1.安装logstash

1. # yum源elasticsearch.repo中就已经包含logstash了，所以无需配置yum源,直接运行以下命令
2. yum install logstash

注意：

如果你的JDK采用源码安装，那么rpm会无法找到你的JAVA_HOME，也就无法配置启动文件，根据我的环境，Centos6.5，我在其他机器上，采用yum安装JDK，yum安装logstash，将生成的启动文件拷贝过来，然后修改一下logstash启动文件中的logstash.lib.sh

1. vim ${logstash_basedir}/bin/logstash.lib.sh
2. export JAVA_HOME=/usr/local/jdk-1.8.0

2.配置logstash

logstash 默认配置文件在/etc/logstash/logstash.yml,通常情况下，不需要做一些改动。

3.启动logstash

==官方给出相关启动方式，根据你的系统可以参考==：运行logstash的三种方式

4.测试logstash正常运行

在终端中运行以下命令：

1. ${logstash_basedir}/bin/logstash -e 'input{stdin{}}output{stdout{codec=>rubydebug}}'

命令执行后，等待你的输入，没问题，输入hello world,回车看看返回什么结果！

1. {
2. "message" => "Hello World",
3. "@version" => "1",
4. "@timestamp" => "2014-08-07T10:30:59.937Z",
5. "host" => "raochenlindeMacBook-Air.local",
6. }

如果返回类似以上内容，没错你已经安装OK了，下面只需要根据你的需求进行日志收集了，这里举一个例子，使用logstash收集nginx access日志。

4.配置logstash收集nginx日志

1. input {
2. path => [ "/usr/local/nginx/logs/access.log" ]
3. type => "nginx_access"
4. }
5. filter {
6. # 这里使用正则过滤，对日志进行清洗,这里就先不详细展示了
7. }
8. output {
9. elasticsearch {
10. hosts => [ "10.10.10.202:9200" ]
11. index => "logstash-%{type}-%{+YYYY.MM.dd}"
12. manage_template => true
13. flush_size => 50000
14. idle_flush_time => 10
15. workers => 2
16. }
17. stdout {codec => rubydebug }
18. }

好了，到这里logstash的基本安装配置就OK了，有关logstash的使用方法，请参考官网：logstash官方文档

七、Filebeat

Filebeat也是elasticsearch的产品，并且软件也包含在elasticsearch YUM源中，可以直接安装

1.安装filebeat

1. yum install -y filebeat

2.配置filbeat

RPM安装

1. ```bash
2. ###################### Filebeat Configuration Example #########################
4. # This file is an example configuration file highlighting only the most common
5. # options. The filebeat.full.yml file from the same directory contains all the
6. # supported options with more comments. You can use it as a reference.
7. #
8. # You can find the full configuration reference here:
9. # https://www.elastic.co/guide/en/beats/filebeat/index.html
11. #=========================== Filebeat prospectors =============================
13. filebeat.prospectors:
15. # Each - is a prospector. Most options can be set at the prospector level, so
16. # you can use different prospectors for various configurations.
17. # Below are the prospector specific configurations.
18. # 指定文件的输入类型log(默认)或者stdin。
19. - input_type: log
20. # paths 指定要监控的日志，可以指定具体得文件或者目录
21. paths:
22. - /var/log/*.log
24. document_type: syslog
25. #- c:programdataelasticsearchlogs*
27. # Exclude lines. A list of regular expressions to match. It drops the lines that are
28. # matching any regular expression from the list.
29. # 在输入中排除符合正则表达式列表的那些行。
30. exclude_lines: ["^DBG"]
32. # Include lines. A list of regular expressions to match. It exports the lines that are
33. # matching any regular expression from the list.
34. # 包含输入中符合正则表达式列表的那些行（默认包含所有行），include_lines执行完毕之后会执行exclude_lines
35. include_lines: ["^ERR", "^WARN"]
37. # Exclude files. A list of regular expressions to match. Filebeat drops the files that
38. # are matching any regular expression from the list. By default, no files are dropped.
39. # 忽略掉符合正则表达式列表的文件
40. #exclude_files: [".gz#34;]
42. # Optional additional fields. These field can be freely picked
43. # to add additional information to the crawled log files for filtering
44. # 向输出的每一条日志添加额外的信息，比如“level:debug”，方便后续对日志进行分组统计。
45. # 默认情况下，会在输出信息的fields子目录下以指定的新增fields建立子目录，例如fields.level
46. # 这个得意思就是会在es中多添加一个字段，格式为 "filelds":{"level":"debug"}
47. #fields:
48. # level: debug
49. # review: 1
51. ### Multiline options
53. # Mutiline can be used for log messages spanning multiple lines. This is common
54. # for Java Stack Traces or C-Line Continuation
55. # 适用于日志中每一条日志占据多行的情况，比如各种语言的报错信息调用栈
57. # The regexp Pattern that has to be matched. The example pattern matches all lines starting with [
58. # 多行日志开始的那一行匹配的pattern
59. #multiline.pattern: ^[
61. # Defines if the pattern set under pattern should be negated or not. Default is false.
62. # 是否需要对pattern条件转置使用，不翻转设为true，反转设置为false。 【建议设置为true】
63. #multiline.negate: false
65. # Match can be set to "after" or "before". It is used to define if lines should be append to a pattern
66. # that was (not) matched before or after or as long as a pattern is not matched based on negate.
67. # Note: After is the equivalent to previous and before is the equivalent to to next in Logstash
68. # 匹配pattern后，与前面（before）还是后面（after）的内容合并为一条日志
69. #multiline.match: after
71. #================================ General =====================================
73. # The name of the shipper that publishes the network data. It can be used to group
74. # all the transactions sent by a single shipper in the web interface.
75. # 用于发布网络数据的shipper名称. 可以被应用于组
76. # 所有的事务通过一个shipper发送到web接口
77. # 默认使用主机名.
78. #name:
80. # The tags of the shipper are included in their own field with each
81. # transaction published.
82. # shipper的标记包含在自己的field中发表事物，通过不同的tags很容易给服务器逻辑分组
83. #tags: ["service-X", "web-tier"]
85. # Optional fields that you can specify to add additional information to the
86. # output.
87. # 可选字段,您可以指定额外的信息添加到输出。字段可以是标量值,数组,字典,或任何的嵌套组合
88. #fields:
89. # env: staging
91. #================================ Outputs =====================================
93. # Configure what outputs to use when sending the data collected by the beat.
94. # Multiple outputs may be used.
96. #-------------------------- Elasticsearch output ------------------------------
97. # 输出到数据配置.单个实例数据可以输出到elasticsearch、logstash、kafka选择其中一种,注释掉其他的输出配置。
98. # 输出数据到Elasticsearch
99. output.elasticsearch:
100. # Array of hosts to connect to.
101. hosts: ["10.10.10.202:9200"]
103. # Optional protocol and basic auth credentials.
104. # 输出认证.
105. #protocol: "https"
106. #username: "elastic"
107. #password: "changeme"
109. #----------------------------- Logstash output --------------------------------
110. output.logstash:
111. # The Logstash hosts
112. # 输出到logstash，由logstash转发给es
113. # 配置logstash的地址，根据个人实际情况而定
114. hosts: ["10.10.10.121:5044"]
116. # Optional SSL. By default is off.
117. # List of root certificates for HTTPS server verifications
118. #ssl.certificate_authorities: ["/etc/pki/root/ca.pem"]
120. # Certificate for SSL client authentication
121. #ssl.certificate: "/etc/pki/client/cert.pem"
123. # Client Certificate Key
124. #ssl.key: "/etc/pki/client/cert.key"
126. #----------------------------- Kafka output ----------------------------------
127. output.kafka:
128. # initial brokers for reading cluster metadata
129. # 输出到kafka，由kafka做缓存，然后传输给logstash，配合logstash_to_es.conf
130. # 配置kafka集群地址
131. hosts: ["10.10.10.121:9092", "10.10.10.123:9092", "10.10.10.125:9092"]
133. # message topic selection + partitioning
134. topic: '%{[type]}'
135. partition.round_robin:
136. reachable_only: false
138. required_acks: 1
139. compression: gzip
140. max_message_bytes: 1000000
141. #================================ Logging =====================================
143. # Sets log level. The default log level is info.
144. # Available log levels are: critical, error, warning, info, debug
145. #logging.level: debug
147. # At debug level, you can selectively enable logging only for some components.
148. # To enable all selectors use ["*"]. Examples of other selectors are "beat",
149. # "publish", "service".
150. #logging.selectors: ["*"]

这里要强调几个filebeat的使用问题：

1.exclude_lines和include_lines，同时配置，会首先执行include_lines,然后再去执行exclude_lines。他们配置没有先后顺序。2.上面的配置文件，有关输出的部分只是展示一下三种输出方式，输出数据的方向只能选择一种，当你选择一种时，务必要把其他的注释掉。

filebeat的相关配置如上所示,关于它的使用可以参考官方文档：filebeat官方文档

上面就是小居数码小编今天给大家介绍的关于（elk+kafka+filebeat）的全部内容，希望可以帮助到你，想了解更多关于数码知识的问题，欢迎关注我们，并收藏，转发，分享。

94%的朋友还想知道的：

• 常见的erp系统有哪些介绍（ERP系统有哪些品牌）
• 360系统重装大师如何使用图解（360系统重装大师怎么使用）
• 录屏软件和iPhone系统自带录屏的区别（苹果录屏软件是什么）
• u盘启动盘重装系统教程（怎样使用u盘重装系统）

（466）个朋友认为回复得到帮助。