本文主要是介绍ELK+kafka+filebeat企业内部日志分析系统，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！
目录
ELK+kafka+filebeat企业内部日志分析系统
1、组件介绍
1、Elasticsearch：
2、Logstash:
3、Kibana:
4、Kafka：
5、Filebeat:
2、环境介绍
3、版本说明
4、搭建架构
5、实施部署
1、 Elasticsearch集群部署
1、安装配置jdk
2、安装配置ES
（1）创建运行ES的普通用户
（2）安装配置ES
（3）设置JVM堆大小 #7.0默认为4G
（4）创建ES数据及日志存储目录
（5）修改安装目录及存储目录权限
3、系统优化
（1）增加最大文件打开数
（2）增加最大进程数
（3）增加最大内存映射数
4、启动ES
5.安装配置head监控插件 （只在第一台es部署）
（1）安装node
（2）下载head插件
（3）安装grunt
（4）修改head源码
（5）下载head必要的文件
（6）运行head
（7）测试
2、 Kibana部署
1. 安装配置Kibana
（1）安装
（2）配置
（3）启动
2. 安装配置Nginx反向代理
（1）配置YUM源：
（2）安装：
（3）配置反向代理
（4）配置授权用户和密码
（5）启动nginx
3、 Kafka部署
1.安装配置jdk8
（1）Kafka、Zookeeper（简称：ZK）运行依赖jdk8
2.安装配置ZK
（1）安装
（2）配置
3.配置Kafka
（1）配置
4、其他kafka节点配置
5、启动、验证ZK集群
（1）启动
（2）验证
6、启动、验证Kafka
（1）启动
（2）验证
7、监控 Kafka Manager
4、 Logstash部署
1.安装配置Logstash
（1）安装
（2）测试文件
（3）配置
5、Filebeat 部署
（1）下载
（2）解压
（3）修改配置
（4）启动
（5）配置nginx
ELK+kafka+filebeat企业内部日志分析系统
1、组件介绍
1、Elasticsearch：
是一个基于Lucene的搜索服务器。提供搜集、分析、存储数据三大功能。它提供了一个分布式多用户能力的全文搜索引擎，基于RESTful web接口。Elasticsearch是用Java开发的，并作为Apache许可条款下的开放源码发布，是当前流行的企业级搜索引擎。设计用于云计算中，能够达到实时搜索，稳定，可靠，快速，安装使用方便。
2、Logstash:
主要是用来日志的搜集、分析、过滤日志的工具。用于管理日志和事件的工具，你可以用它去收集日志、转换日志、解析日志并将他们作为数据提供给其它模块调用，例如搜索、存储等。
3、Kibana:
是一个优秀的前端日志展示框架，它可以非常详细的将日志转化为各种图表，为用户提供强大的数据可视化支持,它能够搜索、展示存储在 Elasticsearch 中索引数据。使用它可以很方便的用图表、表格、地图展示和分析数据。
4、Kafka：
数据缓冲队列。作为消息队列解耦合处理过程，同时提高了可扩展性。具有峰值处理能力，使用消息队列能够使关键组件顶住突发的访问压力，而不会因为突发的超负荷的请求而完全崩溃。
-
1.发布和订阅记录流，类似于消息队列或企业消息传递系统。
-
2.以容错持久的方式存储记录流。
-
3.处理记录发生的流。
5、Filebeat:
隶属于Beats,轻量级数据收集引擎。基于原先 Logstash-fowarder 的源码改造出来。换句话说：Filebeat就是新版的 Logstash-fowarder，也会是 ELK Stack 在 Agent 的第一选择,目前Beats包含四种工具：
-
1.Packetbeat（搜集网络流量数据）
-
2.Metricbeat（搜集系统、进程和文件系统级别的 CPU 和内存使用情况等数据。通过从操作系统和服务收集指标，帮助您监控服务器及其托管的服务。）
-
3.Filebeat（搜集文件数据）
-
4.Winlogbeat（搜集 Windows 事件日志数据）
2、环境介绍
注：以下为环境所需所有服务器，配置为测试环境配置。
| 安装软件 | 主机名 | IP地址 | 系统版本 | 配置 |
|---|---|---|---|---|
| Elasticsearch/Logstash/kibana | Elk | 10.3.145.14 | centos7.5.1804 | 2核4G |
| Elasticsearch | Es1 | 10.3.145.56 | centos7.5.1804 | 2核3G |
| Elasticsearch | Es2 | 10.3.145.57 | centos7.5.1804 | 2核3G |
| zookeeper/kafka | Kafka1 | 10.3.145.41 | centos7.5.1804 | 1核2G |
| zookeeper/kafka | Kafka2 | 10.3.145.42 | centos7.5.1804 | 1核2G |
| zookeeper/kafka | Kafka3 | 10.3.145.43 | centos7.5.1804 | 1核2G |
| Filebeat |
3、版本说明
Elasticsearch: 7.13.2Logstash: 7.13.2Kibana: 7.13.2Kafka: 2.11-1Filebeat: 7.13.2相应的版本最好下载对应的插件
4、搭建架构
1、日志数据由filebate进行收集，定义日志位置，定义kafka集群，定义要传给kafka的那个topic2、kafka接受到数据后,端口为9092，等待消费3、logstash消费kafka中的数据，对数据进行搜集、分析，根据输入条件，过滤条件，输出条件处理后，将数据传输给es集群4、es集群接受数据后，搜集、分析、存储5、kibana提供可视化服务，将es中的数据展示。
相关地址：
官网地址：https://www.elastic.co
官网搭建：Elastic documentation | Elastic
5、实施部署
1、 Elasticsearch集群部署
-
服务器
| 安装软件 | 主机名 | IP地址 | 系统版本 | 配置 |
|---|---|---|---|---|
| Elasticsearch | Elk | 10.3.145.14 | centos7.5.1804 | 2核4G |
| Elasticsearch | Es1 | 10.3.145.56 | centos7.5.1804 | 2核3G |
| Elasticsearch | Es2 | 10.3.145.57 | centos7.5.1804 | 2核3G |
-
软件版本：elasticsearch-7.13.2.tar.gz
-
示例节点：10.3.145.14
1、安装配置jdk
可以自行安装，es安装包中自带了jdk
2、安装配置ES
（1）创建运行ES的普通用户
[root@elk ~]# useradd es[root@elk ~]# echo "******" | passwd --stdin "es"
（2）安装配置ES
[root@elk ~]# tar zxvf /usr/local/package/elasticsearch-7.13.2-linux-x86_64.tar.gz -C /usr/local/[root@elk ~]# mv /usr/local/elasticsearch-7.13.2 /usr/local/es[root@elk ~]# vim /usr/local/es/config/elasticsearch.ymlcluster.name: bjbpe01-elkcluster.initial_master_nodes: ["10.3.145.14","10.3.145.56","10.3.145.57"] # 单节点模式这里的地址只填写本机地址node.name: elk01node.master: truenode.data: truepath.data: /data/elasticsearch/datapath.logs: /data/elasticsearch/logsbootstrap.memory_lock: falsebootstrap.system_call_filter: falsenetwork.host: 0.0.0.0http.port: 9200transport.tcp.port: 9300# 单节点模式下，将discovery开头的行注释discovery.seed_hosts: ["10.3.145.56","10.3.145.57"]discovery.zen.minimum_master_nodes: 2discovery.zen.ping_timeout: 150sdiscovery.zen.fd.ping_retries: 10client.transport.ping_timeout: 60shttp.cors.enabled: truehttp.cors.allow-origin: "*"# 由于我们的笔记本性能有限，如果要使用单节点多实例的话，添加在原有配置中添加node.max_local_storage_nodes: 这个配置限制了单节点上可以开启的ES存储实例的个数
配置项含义：
cluster.name 集群名称，各节点配成相同的集群名称。cluster.initial_master_nodes 集群ip，默认为空，如果为空则加入现有集群，第一次需配置node.name 节点名称，各节点配置不同。node.master 指示某个节点是否符合成为主节点的条件。node.data 指示节点是否为数据节点。数据节点包含并管理索引的一部分。path.data 数据存储目录。path.logs 日志存储目录。bootstrap.memory_lock 内存锁定，是否禁用交换，测试环境建议改为false。bootstrap.system_call_filter 系统调用过滤器。network.host 绑定节点IP。http.port rest api端口。discovery.seed_hosts 提供其他 Elasticsearch 服务节点的单点广播发现功能，这里填写除了本机的其他ipdiscovery.zen.minimum_master_nodes 集群中可工作的具有Master节点资格的最小数量，官方的推荐值是(N/2)+1，其中N是具有master资格的节点的数量。discovery.zen.ping_timeout 节点在发现过程中的等待时间。discovery.zen.fd.ping_retries 节点发现重试次数。http.cors.enabled 是否允许跨源 REST 请求，用于允许head插件访问ES。http.cors.allow-origin 允许的源地址。
（3）设置JVM堆大小 #7.0默认为4G
[root@elk ~]# sed -i 's/## -Xms4g/-Xms4g/' /usr/local/es/config/jvm.options [root@elk ~]# sed -i 's/## -Xmx4g/-Xmx4g/' /usr/local/es/config/jvm.options
注意： 确保堆内存最小值（Xms）与最大值（Xmx）的大小相同，防止程序在运行时改变堆内存大小。 如果系统内存足够大，将堆内存最大和最小值设置为31G，因为有一个32G性能瓶颈问题。 堆内存大小不要超过系统内存的50%
（4）创建ES数据及日志存储目录
[root@elk ~]# mkdir -p /data/elasticsearch/data (/data/elasticsearch)[root@elk ~]# mkdir -p /data/elasticsearch/logs (/log/elasticsearch)
（5）修改安装目录及存储目录权限
[root@elk ~]# chown -R es.es /data/elasticsearch[root@elk ~]# m
3、系统优化
（1）增加最大文件打开数
永久生效方法：
[root@elk ~]# echo "* soft nofile 65536" >> /etc/security/limits.conf
（2）增加最大进程数
[root@elk ~]# echo "* soft nproc 65536" >> /etc/security/limits.conf
* soft nofile 65536* hard nofile 131072* soft nproc 4096* hard nproc 4096更多的参数调整可以直接用这个
（3）增加最大内存映射数
[root@elk ~]# echo "vm.max_map_count=262144" >> /etc/sysctl.conf[root@elk ~]# sysctl -p
启动如果报下列错误
memory locking requested for elasticsearch process but memory is not lockedelasticsearch.yml文件bootstrap.memory_lock : false/etc/sysctl.conf文件vm.swappiness=0错误:max file descriptors [4096] for elasticsearch process is too low, increase to at least [65536]意思是elasticsearch用户拥有的客串建文件描述的权限太低，知道需要65536个解决：切换到root用户下面，[root@elk ~]# vim /etc/security/limits.conf在最后添加* hard nofile 65536* hard nofile 65536重新启动elasticsearch，还是无效？必须重新登录启动elasticsearch的账户才可以，例如我的账户名是elasticsearch，退出重新登录。另外*也可以换为启动elasticsearch的账户也可以，* 代表所有，其实比较不合适启动还会遇到另外一个问题，就是max virtual memory areas vm.max_map_count [65530] is too low, increase to at least [262144]意思是：elasticsearch用户拥有的内存权限太小了，至少需要262114。这个比较简单，也不需要重启，直接执行sysctl -w vm.max_map_count=262144就可以了
4、启动ES
[root@elk ~]# su - es -c "cd /usr/local/es && nohup bin/elasticsearch &"
测试：浏览器访问http://10.3.145.14:9200
5.安装配置head监控插件 （只在第一台es部署）
-
服务器
| 安装软件 | 主机名 | IP地址 | 系统版本 | 配置 |
|---|---|---|---|---|
| Elasticsearch-head-master | Elk | 10.3.145.14 | centos7.5.1804 | 2核4G |
（1）安装node
[root@elk ~]# wget https://npm.taobao.org/mirrors/node/latest-v10.x/node-v10.0.0-linux-x64.tar.gz[root@elk ~]# tar -zxf node-v10.0.0-linux-x64.tar.gz –C /usr/local[root@elk ~]# echo "NODE_HOME=/usr/local/node-v10.0.0-linux-x64PATH=\$NODE_HOME/bin:\$PATHexport NODE_HOME PATH" >>/etc/profile[root@elk ~]# source /etc/profile[root@elk ~]# node --version #检查node版本号
（2）下载head插件
[root@elk ~]# wget https://github.com/mobz/elasticsearch-head/archive/master.zip[root@elk ~]# unzip –d /usr/local elasticsearch-head-master.zip
（3）安装grunt
[root@elk ~]# cd /usr/local/elasticsearch-head-master[root@elk elasticsearch-head-master]# npm install -g grunt-cli[root@elk elasticsearch-head-master]# grunt -version #检查grunt版本号
（4）修改head源码
[root@elk ~]# vi /usr/local/elasticsearch-head-master/Gruntfile.js +99
添加hostname，注意在上一行末尾添加逗号,hostname 不需要添加逗号
[root@elk ~]# vim /usr/local/elasticsearch-head-master/_site/app.js +4388
原本是http://localhost:9200，如果head和ES不在同一个节点，注意修改成ES的IP地址
（5）下载head必要的文件
[root@elk ~]# wget https://github.com/Medium/phantomjs/releases/download/v2.1.1/phantomjs-2.1.1-linux-x86_64.tar.bz2[root@elk ~]# yum -y install bzip2[root@elk ~]# mkdir /tmp/phantomjs[root@elk ~]# mv phantomjs-2.1.1-linux-x86_64.tar.bz2 /tmp/phantomjs/[root@elk ~]# chmod 777 /tmp/phantomjs -R
（6）运行head
[root@elk ~]# cd /usr/local/elasticsearch-head-master/[root@elk elasticsearch-head-master]# npm install [root@elk elasticsearch-head-master]# nohup grunt server &[root@elk elasticsearch-head-master]# ss -tnlpnpm install 执行错误解析：npm ERR! code ELIFECYCLEnpm ERR! errno 1npm ERR! phantomjs-prebuilt@2.1.16 install: `node install.js`npm ERR! Exit status 1npm ERR! npm ERR! Failed at the phantomjs-prebuilt@2.1.16 install script.npm ERR! This is probably not a problem with npm. There is likely additional logging output above.npm ERR! A complete log of this run can be found in:npm ERR! /root/.npm/_logs/2021-04-21T09_49_34_207Z-debug.log解决：[root@elk elasticsearch-head-master]# npm install phantomjs-prebuilt@2.1.16 --ignore-scripts # 具体的版本按照上述报错修改
（7）测试
访问http://10.3.145.14:9100
2、 Kibana部署
-
服务器
| 安装软件 | 主机名 | IP地址 | 系统版本 | 配置 |
|---|---|---|---|---|
| Kibana | Elk | 10.3.145.14 | centos7.5.1804 | 2核4G |
| 软件版本：nginx-1.14.2、kibana-7.13.2-linux-x86_64.tar.gz |
1. 安装配置Kibana
（1）安装
[root@elk ~]# tar zxf kibana-7.13.2-linux-x86_64.tar.gz -C /usr/local/
（2）配置
[root@elk ~]# echo 'server.port: 5601server.host: "10.3.145.14"elasticsearch.hosts: ["http://10.3.145.14:9200"]kibana.index: ".kibana"i18n.locale: "zh-CN"'>>/usr/local/kibana-7.13.2-linux-x86_64/config/kibana.yml
配置项含义：
server.port kibana服务端口，默认5601server.host kibana主机IP地址，默认localhostelasticsearch.url 用来做查询的ES节点的URL，默认http://localhost:9200kibana.index kibana在Elasticsearch中使用索引来存储保存的searches, visualizations和dashboards，默认.kibana
（3）启动
[root@elk ~]# cd /usr/local/kibana-7.13.2-linux-x86_64/[root@elk ~]# nohup ./bin/kibana &
2. 安装配置Nginx反向代理
（1）配置YUM源：
[root@elk ~]# rpm -ivh <http://nginx.org/packages/centos/7/noarch/RPMS/nginx-release-centos-7-0.el7.ngx.noarch.rpm>
（2）安装：
[root@elk ~]# yum install -y nginx httpd-tools 注意：httpd-tools用于生成nginx认证访问的用户密码文件
（3）配置反向代理
[root@elk ~]# cat /etc/nginx/nginx.confuser nginx;worker_processes 4;error_log /var/log/nginx/error.log;pid /var/run/nginx.pid;worker_rlimit_nofile 65535;events {worker_connections 65535;use epoll;}http {include mime.types;default_type application/octet-stream;log_format main '$remote_addr - $remote_user [$time_local] "$request" ''$status $body_bytes_sent "$http_referer" ''"$http_user_agent" "$http_x_forwarded_for"';access_log /var/log/nginx/access.log main;server_names_hash_bucket_size 128;autoindex on;sendfile on;tcp_nopush on;tcp_nodelay on;keepalive_timeout 120;fastcgi_connect_timeout 300;fastcgi_send_timeout 300;fastcgi_read_timeout 300;fastcgi_buffer_size 64k;fastcgi_buffers 4 64k;fastcgi_busy_buffers_size 128k;fastcgi_temp_file_write_size 128k;#gzip模块设置gzip on; #开启gzip压缩输出gzip_min_length 1k; #最小压缩文件大小gzip_buffers 4 16k; #压缩缓冲区gzip_http_version 1.0; #压缩版本（默认1.1，前端如果是squid2.5请使用1.0）gzip_comp_level 2; #压缩等级gzip_types text/plain application/x-javascript text/css application/xml; #压缩类型，默认就已经包含textml，所以下面就不用再写了，写上去也不会有问题，但是会有一个warn。gzip_vary on;#开启限制IP连接数的时候需要使用#limit_zone crawler $binary_remote_addr 10m;#tips:#upstream bakend{#定义负载均衡设备的Ip及设备状态}{# ip_hash;# server 127.0.0.1:9090 down;# server 127.0.0.1:8080 weight=2;# server 127.0.0.1:6060;# server 127.0.0.1:7070 backup;#}#在需要使用负载均衡的server中增加 proxy_pass http://bakend/;server {listen 80;server_name 172.16.244.28;#charset koi8-r;# access_log /var/log/nginx/host.access.log main;access_log off;location / { auth_basic "Kibana"; #可以是string或off，任意string表示开启认证，off表示关闭认证。auth_basic_user_file /etc/nginx/passwd.db; #指定存储用户名和密码的认证文件。proxy_pass http://172.16.244.28:5601;proxy_set_header Host $host:5601; proxy_set_header X-Real-IP $remote_addr; proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for; proxy_set_header Via "nginx"; }location /status { stub_status on; #开启网站监控状态 access_log /var/log/nginx/kibana_status.log; #监控日志 auth_basic "NginxStatus"; } location /head/{auth_basic "head";auth_basic_user_file /etc/nginx/passwd.db;proxy_pass http://172.16.244.25:9100/;proxy_set_header Host $host:9100;proxy_set_header X-Real-IP $remote_addr;proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;proxy_set_header Via "nginx";} # redirect server error pages to the static page /50x.htmlerror_page 500 502 503 504 /50x.html;location = /50x.html {root html;}}}
（4）配置授权用户和密码
[root@elk ~]# htpasswd -cm /etc/nginx/passwd.db kibana
（5）启动nginx
[root@elk ~]# systemctl start nginx
浏览器访问http://10.3.145.14 刚开始没有任何数据，会提示你创建新的索引。
3、 Kafka部署
-
服务器
| 安装软件 | 主机名 | IP地址 | 系统版本 | 配置 |
|---|---|---|---|---|
| zookeeper/kafka | Kafka1 | 10.3.145.41 | centos7.5.1804 | 1核2G |
| zookeeper/kafka | Kafka2 | 10.3.145.42 | centos7.5.1804 | 1核2G |
| zookeeper/kafka | Kafka3 | 10.3.145.43 | centos7.5.1804 | 1核2G |
-
软件版本：jdk-8u121-linux-x64.tar.gz、kafka_2.11-2.0.0.tgz
-
示例节点：10.3.145.41
1.安装配置jdk8
（1）Kafka、Zookeeper（简称：ZK）运行依赖jdk8
[root@kafka1 ~]# tar zxvf /usr/local/package/jdk-8u121-linux-x64.tar.gz -C /usr/local/[root@kafka1 ~]# echo 'JAVA_HOME=/usr/local/jdk1.8.0_121PATH=$JAVA_HOME/bin:$PATHexport JAVA_HOME PATH' >>/etc/profile[root@kafka1 ~]# source /etc/profile
2.安装配置ZK
Kafka运行依赖ZK，Kafka官网提供的tar包中，已经包含了ZK，这里不再额下载ZK程序。
（1）安装
[root@kafka1 ~]# tar zxvf /usr/local/package/kafka_2.11-2.0.0.tgz -C /usr/local/
（2）配置
[root@kafka1 ~]# echo 'dataDir=/opt/data/zookeeper/data dataLogDir=/opt/data/zookeeper/logsclientPort=2181 tickTime=2000 initLimit=20 syncLimit=10 server.1=10.3.145.41:2888:3888 //kafka集群IP:Port .1为id 3处要对应server.2=10.3.145.42:2888:3888server.3=10.3.145.43:2888:3888'> /usr/local/kafka_2.11-2.0.0/config/zookeeper.properties
配置项含义：
dataDir ZK数据存放目录。dataLogDir ZK日志存放目录。clientPort 客户端连接ZK服务的端口。tickTime ZK服务器之间或客户端与服务器之间维持心跳的时间间隔。initLimit 允许follower(相对于Leaderer言的“客户端”)连接并同步到Leader的初始化连接时间，以tickTime为单位。当初始化连接时间超过该值，则表示连接失败。syncLimit Leader与Follower之间发送消息时，请求和应答时间长度。如果follower在设置时间内不能与leader通信，那么此follower将会被丢弃。server.1=172.16.244.31:2888:3888 2888是follower与leader交换信息的端口，3888是当leader挂了时用来执行选举时服务器相互通信的端口。
创建data、log目录[root@kafka1 ~]# mkdir -p /opt/data/zookeeper/{data,logs}创建myid文件[root@kafka1 ~]# echo 1 > /opt/data/zookeeper/data/myid
3.配置Kafka
（1）配置
[root@kafka1 ~]# echo 'broker.id=1listeners=PLAINTEXT://10.3.145.41:9092num.network.threads=3num.io.threads=8socket.send.buffer.bytes=102400socket.receive.buffer.bytes=102400socket.request.max.bytes=104857600log.dirs=/opt/data/kafka/logsnum.partitions=6num.recovery.threads.per.data.dir=1offsets.topic.replication.factor=2transaction.state.log.replication.factor=1transaction.state.log.min.isr=1log.retention.hours=168log.segment.bytes=536870912log.retention.check.interval.ms=300000zookeeper.connect=10.3.145.41:2181,10.3.145.42:2181,10.3.145.43:2181zookeeper.connection.timeout.ms=6000group.initial.rebalance.delay.ms=0' >/usr/local/kafka_2.11-2.0.0/config/server.properties
配置项含义：
broker.id 每个server需要单独配置broker id，如果不配置系统会自动配置。listeners 监听地址，格式PLAINTEXT://IP:端口。num.network.threads 接收和发送网络信息的线程数。num.io.threads 服务器用于处理请求的线程数，其中可能包括磁盘I/O。socket.send.buffer.bytes 套接字服务器使用的发送缓冲区(SO_SNDBUF)socket.receive.buffer.bytes 套接字服务器使用的接收缓冲区(SO_RCVBUF)socket.request.max.bytes 套接字服务器将接受的请求的最大大小(防止OOM)log.dirs 日志文件目录。num.partitions partition数量。num.recovery.threads.per.data.dir 在启动时恢复日志、关闭时刷盘日志每个数据目录的线程的数量，默认1。offsets.topic.replication.factor 偏移量话题的复制因子（设置更高保证可用），为了保证有效的复制，偏移话题的复制因子是可配置的，在偏移话题的第一次请求的时候可用的broker的数量至少为复制因子的大小，否则要么话题创建失败，要么复制因子取可用broker的数量和配置复制因子的最小值。log.retention.hours 日志文件删除之前保留的时间（单位小时），默认168log.segment.bytes 单个日志文件的大小，默认1073741824log.retention.check.interval.ms 检查日志段以查看是否可以根据保留策略删除它们的时间间隔。zookeeper.connect ZK主机地址，如果zookeeper是集群则以逗号隔开。zookeeper.connection.timeout.ms 连接到Zookeeper的超时时间。
创建log目录[root@kafka1 ~]# mkdir -p /opt/data/kafka/logs
4、其他kafka节点配置
只需把配置好的安装包直接分发到其他节点，然后修改ZK的myid，Kafka的broker.id和listeners就可以了。
5、启动、验证ZK集群
（1）启动
在三个节点依次执行：
[root@kafka1 ~]# cd /usr/local/kafka_2.11-2.0.0/[root@kafka1 ~]# nohup bin/zookeeper-server-start.sh config/zookeeper.properties &
（2）验证
查看ZK配置
下载nmap[root@kafka1 ~]# yum install nmap[root@kafka1 ~]# echo conf | nc 127.0.0.1 2181clientPort=2181dataDir=/opt/data/zookeeper/data/version-2dataLogDir=/opt/data/zookeeper/logs/version-2tickTime=2000maxClientCnxns=60minSessionTimeout=4000maxSessionTimeout=40000serverId=1initLimit=20syncLimit=10electionAlg=3electionPort=3888quorumPort=2888peerType=0
查看ZK状态
[root@kafka1 ~]# echo stat |nc 127.0.0.1 2181Zookeeper version: 3.4.13-2d71af4dbe22557fda74f9a9b4309b15a7487f03, built on 06/29/2018 00:39 GMTClients:/127.0.0.1:51876[0](queued=0,recved=1,sent=0)Latency min/avg/max: 0/0/0Received: 2Sent: 1Connections: 1Outstanding: 0Zxid: 0x0Mode: followerNode count: 4
查看端口
[root@kafka1 ~]# lsof -i:2181COMMAND PID USER FD TYPE DEVICE SIZE/OFF NODE NAMEjava 15002 root 98u IPv4 43385 0t0 TCP *:eforward (LISTEN)
6、启动、验证Kafka
（1）启动
在三个节点依次执行：
[root@kafka1 ~]# cd /usr/local/kafka_2.11-2.0.0/[root@kafka1 ~]# nohup bin/kafka-server-start.sh config/server.properties &
（2）验证
在10.3.145.41上创建topic
[root@kafka1 ~]# bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic testtopicCreated topic "testtopic".
查询10.3.145.41上的topic
[root@kafka1 ~]# bin/kafka-topics.sh --zookeeper 10.3.145.41:2181 --list testtopic
查询10.3.145.42上的topic
[root@kafka1 ~]# bin/kafka-topics.sh --zookeeper 10.3.145.42:2181 --list testtopic
查询10.3.145.43上的topic
[root@kafka1 ~]# bin/kafka-topics.sh --zookeeper 10.3.145.43:2181 --list testtopic
模拟消息生产和消费 发送消息到10.3.145.41
[root@kafka1 ~]# bin/kafka-console-producer.sh --broker-list 10.3.145.41:9092 --topic testtopic >Hello World!
从10.3.145.42接受消息
[root@kafka1 ~]# bin/kafka-console-consumer.sh --bootstrap-server 10.3.145.41:9092 --topic testtopic --from-beginning Hello World!
7、监控 Kafka Manager
Kafka-manager 是 Yahoo 公司开源的集群管理工具。
可以在 Github 上下载安装：GitHub - yahoo/CMAK: CMAK is a tool for managing Apache Kafka clusters
如果遇到 Kafka 消费不及时的话，可以通过到具体 cluster 页面上，增加 partition。Kafka 通过 partition 分区来提高并发消费速度
4、 Logstash部署
-
服务器
| 安装软件 | 主机名 | IP地址 | 系统版本 | 配置 |
|---|---|---|---|---|
| Logstash | Elk | 10.3.145.14 | centos7.5.1804 | 2核4G |
-
软件版本：logstash-7.13.2.tar.gz
1.安装配置Logstash
Logstash运行同样依赖jdk，本次为节省资源，故将Logstash安装在了10.3.145.14节点。
（1）安装
[root@elk ~]# tar zxf /usr/local/package/logstash-7.13.2.tar.gz -C /usr/local/
（2）测试文件
标准输入=>标准输出
1、启动logstash
2、logstash启动后，直接进行数据输入
3、logstash处理后，直接进行返回
input {stdin {}}output {stdout {codec => rubydebug}}
标准输入=>标准输出及es集群
1、启动logstash
2、启动后直接在终端输入数据
3、数据会由logstash处理后返回并存储到es集群中
input {stdin {}}output {stdout {codec => rubydebug}elasticsearch {hosts => ["10.3.145.14","10.3.145.56","10.3.145.57"]index => 'logstash-debug（zu）-%{+YYYY-MM-dd}'}}
端口输入=>字段匹配=>标准输出及es集群
1、由tcp 的8888端口将日志发送到logstash
2、数据被grok进行正则匹配处理
3、处理后，数据将被打印到终端并存储到es
input {tcp {port => 8888}}filter {grok {match => {"message" => "%{DATA:key} %{NUMBER:value:int}"} }}output {stdout {codec => rubydebug}elasticsearch {hosts => ["10.3.145.14","10.3.145.56","10.3.145.57"]index => 'logstash-debug-%{+YYYY-MM-dd}'}}# yum install -y nc# free -m |awk 'NF==2{print $1,$3}' |nc logstash_ip 8888
文件输入=>字段匹配及修改时间格式修改=>es集群
1、直接将本地的日志数据拉去到logstash当中
2、将日志进行处理后存储到es
input {file {type => "nginx-log"path => "/var/log/nginx/error.log"start_position => "beginning" # 此参数表示在第一次读取日志时从头读取# sincedb_path => "自定义位置" # 此参数记录了读取日志的位置，默认在 data/plugins/inputs/file/.sincedb*}}filter {grok {match => { "message" => '%{DATESTAMP:date} [%{WORD:level}] %{DATA:msg} client: %{IPV4:cip},%{DATA}"%{DATA:url}"%{DATA}"%{IPV4:host}"'} } date {match => [ "timestamp" , "dd/MMM/YYYY:HH:mm:ss Z" ] } }output {if [type] == "nginx-log" {elasticsearch {hosts => ["192.168.249.139:9200","192.168.249.149:9200","192.168.249.159:9200"]index => 'logstash-audit_log-%{+YYYY-MM-dd}'}}}
filebeat => 字段匹配 => 标准输出及es
input {beats {port => 5000}}filter {grok {match => {"message" => "%{IPV4:cip}"} }}output {elasticsearch {hosts => ["192.168.249.139:9200","192.168.249.149:9200","192.168.249.159:9200"]index => 'test-%{+YYYY-MM-dd}'}stdout { codec => rubydebug }}
（3）配置
创建目录，我们将所有input、filter、output配置文件全部放到该目录中。
[root@elk ~]# mkdir -p /usr/local/logstash-7.13.2/etc/conf.d[root@elk ~]# vim /usr/local/logstash-7.13.2/etc/conf.d/input.confinput { kafka {type => "audit_log"codec => "json"topics => "nginx"decorate_events => truebootstrap_servers => "10.3.145.41:9092, 10.3.145.42:9092, 10.3.145.43:9092"}}[root@elk ~]# vim /usr/local/logstash-7.13.2/etc/conf.d/filter.conffilter {json { # 如果日志原格式是json的，需要用json插件处理source => "message"target => "nginx" # 组名}}[root@elk ~]# vim /usr/local/logstash-7.13.2/etc/conf.d/output.confoutput {if [type] == "audit_log" {elasticsearch {hosts => ["10.3.145.14","10.3.145.56","10.3.145.57"]index => 'logstash-audit_log-%{+YYYY-MM-dd}'}}}
（3）启动
[root@elk ~]# cd /usr/local/logstash-7.13.2[root@elk ~]# nohup bin/logstash -f etc/conf.d/ --config.reload.automatic &
5、Filebeat 部署
为什么用 Filebeat ，而不用原来的 Logstash 呢？
原因很简单，资源消耗比较大。
由于 Logstash 是跑在 JVM 上面，资源消耗比较大，后来作者用 GO 写了一个功能较少但是资源消耗也小的轻量级的 Agent 叫 Logstash-forwarder。
后来作者加入 elastic.co 公司， Logstash-forwarder 的开发工作给公司内部 GO 团队来搞，最后命名为 Filebeat。
Filebeat 需要部署在每台应用服务器上，可以通过 Salt 来推送并安装配置。
-
服务器
| 安装软件 | 主机名 | IP地址 | 系统版本 | 配置 |
|---|---|---|---|---|
| filebeat | Kafka3 | 10.3.145.43 | centos7.5.1804 | 1核2G |
-
软件版本 filebeat-7.13.2-x86_64.rpm
（1）下载
[root@kafka3 ~]# curl -L -O https://artifacts.elastic.co/downloads/beats/filebeat/filebeat-7.13.2-x86_64.rpm
（2）解压
[root@kafka3 ~]# yum install -y filebeat-7.13.2-x86_64.rpm
（3）修改配置
修改 Filebeat 配置，支持收集本地目录日志，并输出日志到 Kafka 集群中
[root@kafka3 ~]# vim filebeat.ymlfilebeat.inputs:- type: logenabled: truepaths:- /var/log/nginx/access.logoutput.logstash:hosts: ["10.3.145.14:5000"]output.kafka: hosts: ["10.3.145.41:9092","10.3.145.42:9092","10.3.145.43:9092"]topic: 'nginx'# 注意，如果需要重新读取，请删除/data/registry目录
Filebeat 6.0 之后一些配置参数变动比较大，比如 document_type 就不支持，需要用 fields 来代替等等。
（4）启动
[root@kafka3 ~]# ./filebeat -e -c filebeat.yml
（5）配置nginx
因为日志格式的切割需要json格式，kibana中会报错 error decoding json，所以在这里我们将nginx的日志格式修改为json格式。[root@kafka3 ~]# vim /etc/nginx/nginx.conf
# log_format main '$remote_addr - $remote_user [$time_local] "$request" '
# '$status $body_bytes_sent "$http_referer" '
# '"$http_user_agent" "$http_x_forwarded_for"';log_format main '{"user_ip":"$http_x_real_ip","lan_ip":"$remote_addr","log_time":"$time_iso8601","user_req":"$request","http_code":"$status","body_bytes_sents":"$body_bytes_sent","req_time":"$request_time","user_ua":"$http_user_agent"}';access_log /var/log/nginx/access.log main;
这篇关于ELK+kafka+filebeat企业内部日志分析系统的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！
