基于Filebeat、Kafka搭建ELK日志分析平台详细步骤-Toy模板网

这篇具有很好参考价值的文章主要介绍了基于Filebeat、Kafka搭建ELK日志分析平台详细步骤。希望对大家有所帮助。如果存在错误或未考虑完全的地方，请大家不吝赐教，您也可以点击"举报违法"按钮提交疑问。

ELK搭建详细步骤

写在前头：公司一直没有搭建一个支持实时查询日志的平台，平常看日志都得去服务器下载，大大降低开发效率，前段时间有大佬同事搭建了一款日志平台，支持sit,uat等各个环境的日志实时查询，大大提高bug定位速度。因对其感兴趣特向大佬请教，学习记录下搭建流程。

技术选型以及搭建架构

选型

Elasticsearch

Elasticsearch 是一个分布式的 RESTful 风格的搜索和数据分析引擎，提供收集、分析、存储数据三大功能。是当前流行的企业及搜索引擎。设计用于云计算中，能够达到实时搜索，稳定、可靠、快速。

Logstash

Logstash是一个开源数据收集引擎，具有实时管道功能。Logstash可以动态地将来自不同数据源的数据统一起来，并将数据标准化到你所选择的目的地。

Kibana

Kibana 是为 Elasticsearch设计的开源分析和可视化平台。你可以使用 Kibana 来搜索，查看存储在 Elasticsearch 索引中的数据并与之交互。你可以很容易实现高级的数据分析和可视化，以图表的形式展现出来。

Kafka

Kafka是一种高吞吐量的分布式发布订阅消息系统，你懂的，解耦、削峰等等。

Filebeat

首先filebeat是Beats中的一员,Beats在是一个轻量级日志采集器，其实Beats家族有6个成员，早期的ELK架构中使用Logstash收集、解析日志，但是Logstash对内存、cpu、io等资源消耗比较高。相比Logstash，Beats所占系统的CPU和内存几乎可以忽略不计。

架构

filebeat kafka配置详解,kafka,elk,elasticsearch

环境准备

主机

主机名	系统版本	安装软件	配置要求
pihao101	centos7.6	jdk1.8/Elasticsearch/zookeeper/kafka/kibana/nginx	4G
pihao102	centos7.6	jdk1.8/Elasticsearch/zookeeper/kafka/Logstash	4G
pihao103	centos7.6	jdk1.8/Elasticsearch/zookeeper/kafka/filebeat/nginx	4G

软件版本

jdk	1.8
Elasticsearch	6.5.4
Logstash	6.5.4
Kibana	6.5.4
Kafka	2.11-1
Filebeat	6.5.4
Nginx	都行

准备搭建

配置java环境变量

因为elk的大部分工具都需要jdk的依赖

分别在pihao101,pihao102,pihao103上配置java

1、使用ftp工具将jdk上传上 /opt/software目录下

filebeat kafka配置详解,kafka,elk,elasticsearch

2、解压jdk

# 解压jdk
tar -zxvf jdk-8u212-linux-x64.tar.gz -C ../module

3、在 /etc/profile.d/ 目录下创建my_env.sh脚本

sudo touch my_env.sh

#配置JDK的环境变量
# 声明 JAVA_HOME变量
JAVA_HOME=/opt/module/jdk1.8.0_212
# 声明PATH变量,在系统原有PATH变量上追加 $JAVA_HOME/bin
PATH=$PATH:$JAVA_HOME/bin

#提升PATH JAVA_HOME为系统全局变量
export JAVA_HOME PATH

4、使用source命令让环境生效

# 重新加载配置
source /etc/profile
# 检测是否成功
java -version

filebeat kafka配置详解,kafka,elk,elasticsearch

至此，三台机器上的java已经安装好了！开始着手安装其他工具

Elasticsearch集群部署

ES运行需要依赖jdk1.8,上传ES

filebeat kafka配置详解,kafka,elk,elasticsearch

1、创建运行ES的普通用户

ES运行不能以root用户进行，三台都创建

useradd ela
passwd ela

2、安装配置ES

三台机器分别解压

[root@pihao101 software]# tar -zxvf elasticsearch-6.5.4.tar.gz -C ../module/

修改pihao101上的elasticsearch.yml配置文件

cluster.name: pihao-elk            #集群名称
path:
  data: /data/elasticsearch/data   #数据路径
  logs: /data/elasticsearch/logs   #日志路径
node:
  name: elk01                      #节点名称
  master: true
  data: true
bootstrap:
  memory_lock: true         		#内存锁定，是否禁用交换
  system_call_filter: false 		# 系统调用过滤器
network.host: 0.0.0.0       		#环回地址，单节点配置为本机地址，集群环境配置成0.0.0.0
http.port: 9200

#es服务节点发现
discovery.zen.ping.unicast.hosts: [pihao102,pihao103]
#集群中可工作的具有Master节点资格的最小数量
discovery.zen.minimum_master_nodes: 2 
#节点在发现过程中的等待时间
discovery.zen.ping_timeout: 150s  
#节点发现重试次数
discovery.zen.fd.ping_retries: 10 
client.transport.ping_timeout: 60s
#是否允许跨域
http.cors.enabled: true 
#允许的源地址
http.cors.allow-origin: "*"

修改pihao102上的elasticsearch.yml配置文件

node.name: elk02  
discovery.zen.ping.unicast.hosts: [pihao101,pihao103]

修改pihao103上的elasticsearch.yml配置文件

node.name: elk03  
discovery.zen.ping.unicast.hosts: [pihao101,pihao102]

设置JVM堆大小

视情况而定，修改config下面的jvm.options文件

filebeat kafka配置详解,kafka,elk,elasticsearch

创建ES数据以及日志存储文件并修改权限

mkdir -pv /data/elasticsearch/data
mkdir -pv /data/elasticsearch/logs

chown -R ela:ela /data/elasticsearch
chown -R ela:ela /opt/module/elasticsearch-6.5.4 # 安装目录

3、系统修改

vim /etc/security/limits.conf

增加最大文件打开数

增加最大进程数

* soft nofile 65536
* hard nofile 131072
* soft nproc 2048
* hard nproc 4096

增加最大内存映射数

echo "vm.max_map_count=262144" >> /etc/sysctl.conf
sysctl -p

4、启动ES

在启动之前，先将所有的机器都切换到ela用户

su ela

cd /opt/module/elasticsearch-6.5.4

[ela@pihao101 elasticsearch-6.5.4]$ pwd
/opt/module/elasticsearch-6.5.4
[ela@pihao101 elasticsearch-6.5.4]$ ll
总用量 448
drwxr-xr-x.  3 ela ela   4096 3月  11 15:23 bin
drwxr-xr-x.  2 ela ela   4096 3月  11 16:32 config
drwxr-xr-x.  3 ela ela   4096 12月 18 2018 lib
-rw-r--r--.  1 ela ela  13675 12月 18 2018 LICENSE.txt
drwxr-xr-x.  2 ela ela   4096 12月 18 2018 logs
drwxr-xr-x. 28 ela ela   4096 12月 18 2018 modules
-rw-r--r--.  1 ela ela 403816 12月 18 2018 NOTICE.txt
drwxr-xr-x.  2 ela ela   4096 12月 18 2018 plugins
-rw-r--r--.  1 ela ela   8519 12月 18 2018 README.textile

# 三台机器都要启动
[ela@pihao101 elasticsearch-6.5.4]$ nohup bin/elasticsearch &

注意：如果启动的时候报错，“memory locking requested for elasticsearch process but memory is not locked”

filebeat kafka配置详解,kafka,elk,elasticsearch

bootstrap:
  memory_lock: false

修改/etc/sysctl.conf文件

vm.swappiness=0
sysctl -p

启动成功

可以查看日志看看各个节点是否都已经启动成功

也可以访问浏览器

http://pihao101:9200 http://pihao102:9200 http://pihao103:9200

filebeat kafka配置详解,kafka,elk,elasticsearch

访问浏览器

filebeat kafka配置详解,kafka,elk,elasticsearch

好了，至此，ES搭建完毕。接下来准备安装一个ES插件，方便我们可视化的监控ES

安装head监控插件

head插件是nodejs实现的，所以需要先安装node

这里我选择的是node-v4.4.7-linux-x64.tar.gz

该插件只要安装在一个节点就行，不用全部安装

安装node

从ela用户切换会root用户

1、下载

下载地址：https://registry.npmmirror.com/binary.html?path=node/v4.4.7/

# 或者如下命令下载
# wget https://npm.taobao.org/mirrors/node/latest-v4.x/node-v4.4.7-linux-x64.tar.gz

filebeat kafka配置详解,kafka,elk,elasticsearch

2、解压

# tar -zxvf node-v4.4.7-linux-x64.tar.gz -C /opt/module

filebeat kafka配置详解,kafka,elk,elasticsearch

3、配置node环境变量

vim /etc/profile.d/my_env.sh ,配置如下

NODE_HOME=/opt/module/node-v4.4.7-linux-x64
PATH=$NODE_HOME/bin:$PATH
export NODE_HOME PATH

filebeat kafka配置详解,kafka,elk,elasticsearch

4、重新加载环境

source /etc/profile
node --version

filebeat kafka配置详解,kafka,elk,elasticsearch

下载head插件

地址：https://github.com/mobz/elasticsearch-head/archive/master.zip

上传至software并解压

unzip -d ../module/ elasticsearch-head-master.zip

# 小意外，unzip命令找不到，使用yum -y install unzip	太慢
# 更换阿里云的源: https://developer.aliyun.com/mirro
rm -rf /etc/yum.repos.d/*

curl -o /etc/yum.repos.d/CentOS-Base.repo https://mirrors.aliyun.com/repo/Centos-7.repo
yum -y install wget
wget -O /etc/yum.repos.d/epel.repo https://mirrors.aliyun.com/repo/epel-7.repo

# ok，保存源完毕

filebeat kafka配置详解,kafka,elk,elasticsearch

安装grunt

cd /opt/module/elasticsearch-head-master
npm install -g grunt-cli@1.3.2 # 这里指定下版本，版本不用可能存在兼容问题报错
grunt --version

filebeat kafka配置详解,kafka,elk,elasticsearch

修改head源码

# vim /opt/module/elasticsearch-head-master/Gruntfile.js    +95
如下图：

filebeat kafka配置详解,kafka,elk,elasticsearch

# vim /opt/module/elasticsearch-head-master/_site/app.js     +4374
如下图：

filebeat kafka配置详解,kafka,elk,elasticsearch

下载head必要的文件

https://github.com/Medium/phantomjs/releases/download/v2.1.1/phantomjs-2.1.1-linux-x86_64.tar.bz2
上传至software目录
[root@pihao101 software]# mkdir /tmp/phantomjs
[root@pihao101 software]# cp phantomjs-2.1.1-linux-x86_64.tar.bz2 /tmp/phantomjs/

filebeat kafka配置详解,kafka,elk,elasticsearch

运行head

[root@pihao101 phantomjs]# cd /opt/module/elasticsearch-head-master/
npm install --registry=https://registry.npm.taobao.org 
该步骤过程大概持续十分钟

filebeat kafka配置详解,kafka,elk,elasticsearch

# 尝试解决上面的报错信息
yum -y install bzip2.x86_64
npm install --registry=https://registry.npm.taobao.org --unsafe-perm


最后安装完成！
[root@pihao101 phantomjs]# nohup grunt server &

测试

访问http://pihao101:9100

filebeat kafka配置详解,kafka,elk,elasticsearch

Kibana部署

这里我安装到101节点

安装配置Kibana

1、上传并解压

[root@pihao101 software]# pwd
/opt/software
[root@pihao101 software]# tar -zxvf kibana-6.5.4-linux-x86_64.tar.gz -C ../module/

2、配置

修改配置文件 kibana.yml,配置如下

vim config/kibana.yml

server.port: 5601
server.host: pihao101
elasticsearch.url: "http://pihao101:9200"
# kibana在es中使用索引来存储保存的searches，visualizations和dashboard,默认是.kibana
kibana.index: ".kibana"

3、启动

[root@pihao101 kibana-6.5.4-linux-x86_64]# cd /opt/module/kibana-6.5.4-linux-x86_64/
[root@pihao101 kibana-6.5.4-linux-x86_64]# nohup bin/kibana &

filebeat kafka配置详解,kafka,elk,elasticsearch

安装配置nginx

1、配置YUM源

rpm -ivh https://nginx.org/packages/centos/7/noarch/RPMS/nginx-release-centos-7-0.el7.ngx.noarch.rpm

2、安装

[root@pihao101 software]# yum install -y nginx httpd-tools
#查看nginx安装位置
[root@pihao101 software]# whereis nginx
nginx: /etc/nginx

3、启动验证nginx

# 启动服务 restart stop
systemctl start nginx

# nginx的其他命令
nginx
nginx -s reload

filebeat kafka配置详解,kafka,elk,elasticsearch

4、使用http-tools生成用户密码

# httpd-tools是用来生成nginx认证访问的用户密码文件
[root@pihao101 nginx]# yum install -y nginx httpd-tools

# c 创建一个;m md5加密方式; 指定账号
[root@pihao101 nginx]# htpasswd -cm /etc/nginx/passwd.db kibana
New password: 
Re-type new password: 
Adding password for user kibana
[root@pihao101 nginx]# cat /etc/nginx/passwd.db 
kibana:$apr1$b8WAMxNT$M/pnhtcTCmiUcQW61F6yA0
[root@pihao101 nginx]#

5、修改Nginx conf文件配置代理

user  nginx;
worker_processes  auto;

error_log  /var/log/nginx/error.log notice;
pid        /var/run/nginx.pid;
worker_rlimit_nofile 65535;

events {
    worker_connections  65536;
    use epoll;
}


http {
    include       /etc/nginx/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_min_length 1k;
    gzip_buffers 4 16k;
    gzip_http_version 1.0;
    gzip_comp_level 2;
    gzip_types text/plain application/x-javascript text/css application/xml;
    gzip_vary on;
    

	server {
        listen       80;
        server_name  pihao101;

   		#access_log  /var/log/nginx/host.access.log  main;
		access_log off;
		
        location / {
        	auth_basic "Kibana";
        	auth_basic_user_file /etc/nginx/passwd.db; 
        	proxy_pass http://pihao101: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://pihao101: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.html
		error_page   500 502 503 504  /50x.html;
		location = /50x.html {
			root   /usr/share/nginx/html;
		}
    
	}
}

# 检查配置有没有问题
[root@pihao101 nginx]# nginx -t
nginx: the configuration file /etc/nginx/nginx.conf syntax is ok
nginx: configuration file /etc/nginx/nginx.conf test is successful
[root@pihao101 nginx]# 
# 重启nginx
[root@Centos-node6 ~]# systemctl restart nginx

坑！修改好配置的时候，访问192.168.1.101(pihao),显示404，经查看错误日志后发现如下提示：

filebeat kafka配置详解,kafka,elk,elasticsearch

# 解决办法如下：是因为SeLinux的限制。
# 关闭 1为开启
[root@pihao101 nginx]# setenforce 0

接着继续访问 192.168.1.101（pihao101）,发现登录验证配置生效，输入密码后跳转成功！

filebeat kafka配置详解,kafka,elk,elasticsearch

Kafka部署

搭建集群，分别在三个节点部署。

前提条件：由于我这个版本的kafka的启动还需要依赖Zookeeper,所以需要先安装zk的集群，并且还需要jdk1.8环境（之前已安装好）

安装配置zookeeper

1、上传并解压，kafka中有zookeeper的安装包

[root@pihao101 software]# tar -zxvf kafka_2.11-2.0.0.tgz -C ../module/

2、配置

修改/opt/module/kafka_2.11-2.0.0/config/zookeeper.properties

pihao101，102，103节点配置如下：

#zk数据存放目录
dataDir=/opt/data/zookeeper/data
#zk日志存放目录
dataLogDir=/opt/data/zookeeper/logs
clientPort=2181
#zk服务器之间或客户端与服务器之间维持心跳的时间间隔
tickTime=2000
# 允许follower连接并同步到Leader的初始化连接时间
initLimit=20
syncLimit=10
#集群信息 2888是follower与leader交换信息的端口，3888是当leader挂了时用来执行选举时的通信的端口
#这里我开始是用的pihao101,pihao102,pihao103,后来启动报错了就换成了具体的ip
server.1=192.168.1.101:2888:3888
server.2=192.168.1.102:2888:3888
server.3=192.168.1.103:2888:3888

紧接着创建data、log目录

mkdir -p /opt/data/zookeeper/data
mkdir -p /opt/data/zookeeper/logs

创建myid文件

# 确保每台kafka节点的id唯一
# pihao101
echo 1 > /opt/data/zookeeper/data/myid

# pihao102
echo 2 > /opt/data/zookeeper/data/myid

# pihao103
echo 3 > /opt/data/zookeeper/data/myid

安装配置kafka

修改 opt/module/kafka_2.11-2.0.0/config/server.properties

pihao101节点配置如下：

# 每个机器需要单独配置一个broker id
broker.id=1
# 监听地址
listeners=PLAINTEXT://pihao101:9092
# 接收和发送网络信息的线程数
num.network.threads=3
# 服务器用于处理请求的线程数
num.io.threads=8
socket.send.buffer.bytes=102400
socket.receive.buffer.bytes=102400
socket.request.max.bytes=104857600
# 日志文件目录
log.dirs=/opt/data/kafka/logs
num.partitions=6
num.recovery.threads.per.data.dir=1
offsets.topic.replication.factor=2
transaction.state.log.replication.factor=1
transaction.state.log.min.isr=1
log.retention.hours=168
log.segment.bytes=536870912
log.retention.check.interval.ms=300000
zookeeper.connect=pihao101:2181,pihao102:2181,pihao103:2181
zookeeper.connection.timeout.ms=6000
group.initial.rebalance.delay.ms=0

另外节点配置修改如下：

# pihao102
broker.id=2
listeners=PLAINTEXT://pihao102:9092

# pihao103
broker.id=3
listeners=PLAINTEXT://pihao103:9092

创建log文件

# 三台机器
[root@pihao101 config]# mkdir -p /opt/data/kafka/logs

启动、验证ZK集群

1、启动

# 在三个节点分别执行
[root@pihao101 config]# cd /opt/module/kafka_2.11-2.0.0/
[root@pihao101 kafka_2.11-2.0.0]# nohup bin/zookeeper-server-start.sh config/zookeeper.properties &

报错如下：

filebeat kafka配置详解,kafka,elk,elasticsearch

解决办法：

在zookeeper.properties配置中修改如下：

filebeat kafka配置详解,kafka,elk,elasticsearch

2、验证

使用nc来验证，先安装nc

# yum -y install nc

# 查看zk的配置
echo conf | nc 127.0.0.1 2181

filebeat kafka配置详解,kafka,elk,elasticsearch

# 查看zk的状态
echo stat | nc 127.0.0.1 2181

filebeat kafka配置详解,kafka,elk,elasticsearch

# 查看2181端口
# lsof -i:2181
[root@pihao101 kafka_2.11-2.0.0]# lsof -i:2181
COMMAND   PID USER   FD   TYPE DEVICE SIZE/OFF NODE NAME
java    16424 root   96u  IPv6  59318      0t0  TCP *:eforward (LISTEN)

至此，zookeeper集群启动成功，接下来开始启动kafka!

启动、验证kafka

1、启动

在三个节点上分别执行如下命令

[root@pihao101 kafka_2.11-2.0.0]# cd /opt/module/kafka_2.11-2.0.0/
[root@pihao101 kafka_2.11-2.0.0]# nohup bin/kafka-server-start.sh config/server.properties &

2、验证

在pihao101上创建测试的topic

[root@pihao101 kafka_2.11-2.0.0]# bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test1

查询三台机器上的topic

# 101
[root@pihao101 kafka_2.11-2.0.0]# bin/kafka-topics.sh --zookeeper pihao101:2181 --list
test1
# 102
[root@pihao101 kafka_2.11-2.0.0]# bin/kafka-topics.sh --zookeeper pihao102:2181 --list
test1
# 103
[root@pihao101 kafka_2.11-2.0.0]# bin/kafka-topics.sh --zookeeper pihao103:2181 --list
test1

# 查看kafka中topic的数据
bin/kafka-console-consumer.sh --bootstrap-server pihao103:9092 --topic test1 --from-beginning

验证ok，集群应该是正常的

模拟消息的生产和消费

发送消息到101节点

[root@pihao101 kafka_2.11-2.0.0]# bin/kafka-console-producer.sh --broker-list pihao101:9092 --topic test1

从102节点上消费数据

[root@pihao102 kafka_2.11-2.0.0]# bin/kafka-console-consumer.sh --bootstrap-server pihao102:9092 --topic test1 --from-beginning

filebeat kafka配置详解,kafka,elk,elasticsearch

至此，kafka集群搭建完毕！

Logstash部署

按照之前的规划，Logstatsh部署在102节点上，注意，Logstash同样需要依赖jdk的环境

安装配置Logstash

1、安装

[root@pihao102 ~]# cd /opt/software/
[root@pihao102 software]# tar -zxvf logstash-6.5.4.tar.gz -C ../module/

2、配置

提前创建一个目录，将input,filter,output的配置文件全部放到该目录下

[root@pihao102 logstash-6.5.4]# mkdir -p /opt/module/logstash-6.5.4/etc/conf.d

#配置input
# vim /opt/module/logstash-6.5.4/etc/conf.d/input.conf
input {
    kafka {
        type => "nginx_kafka"
        codec => "json"
        topics => "test1"
        decorate_events => true
        bootstrap_servers => "192.168.1.101:9092, 192.168.1.102:9092, 192.168.1.103:9092"
    }
}

#配置output
# vim /opt/module/logstash-6.5.4/etc/conf.d/output.conf
output {
    if[type] == "nginx_kafka" {
        elasticsearch {
            hosts => ["192.168.1.101","192.168.1.102","192.168.1.103"]
            index => 'logstash-test1-%{+YYYY-MM-dd}'
        }
    }
}

3、启动

启动之前可以使用命令检测下配置是否ok

bin/logstash -f etc/conf.d/output.conf -t

坑：之前的output的配置多了 “}”

[root@pihao102 conf.d]# clear
[root@pihao102 conf.d]# cd /opt/module/logstash-6.5.4/
[root@pihao102 logstash-6.5.4]# nohup bin/logstash -f etc/conf.d/ --config.reload.automatic &

Filebeat部署

filebeat，轻量，资源消耗小

按照规划，将Filebeat部署在103上面

1、上传并解压

[root@pihao103 kafka_2.11-2.0.0]# cd /opt/software/
[root@pihao103 software]# tar -zxvf filebeat-6.5.4-linux-x86_64.tar.gz -C ../module/

filebeat kafka配置详解,kafka,elk,elasticsearch

2、修改配置

修改filebeat配置，支持收集本地目录日志，并输出日志到Kafka集群中.

[root@pihao103 filebeat-6.5.4-linux-x86_64]# cd /opt/module/filebeat-6.5.4-linux-x86_64/
[root@pihao103 filebeat-6.5.4-linux-x86_64]# vim filebeat.yml 
# 修改配置如下：

第一次启动，采集的日志文件为pihao103节点上nginx的log

暂时这么配置测试启动用

快速在103节点上搭一个nginx服务器（过程略，参考之前的步骤）

filebeat.prospectors:
- input_type: log
  paths:
  	- /var/log/nginx/access.log
  json.keys_under_root: true
  json.add_error_key: true
  json.message_key: log
  
output.kafka:
	hosts: ["192.168.1.101:9092","192.168.1.102:9092","192.168.1.103:9092"]
	topic: "test1"

配置解释

# 可以让字段位于根节点，默认为false
json.keys_under_root: true	
# 对于同名的key,覆盖原有的key值
json.overwrite_keys: true     
# 用来合并多行json日志使用，如果配置该项还需要配置multiline的设置
json.message_key: message
# 将解析错误的消息存储在error.message字段中
json.add_error_key: true

3、启动

[root@pihao103 kafka_2.11-2.0.0]# cd /opt/module/filebeat-6.5.4-linux-x86_64/
[root@pihao103 filebeat-6.5.4-linux-x86_64]# nohup ./filebeat -e -c filebeat.yml &
[root@pihao103 filebeat-6.5.4-linux-x86_64]# tail -f nohup.out

filebeat kafka配置详解,kafka,elk,elasticsearch

至此，本次ELK的基本环境已经搭建完毕，filebeat --> kafka —>logstash —> es -->kibana环节全部打通

接下来开始演示具体采集流程

ELK日志采集演示

演示流程

filebeat: 采集所在节点上的nginx的日志信息，然后把信息发送到kafka中【test1】topic中
kafka: 接收filebeat采集的数据
logstash:将kafka【test1】topic的数据传输的elasticsearch的【logstash-test1-%{+YYYY-MM-dd}】索引
elasticsearch: 存储logstash中的数据，并传输给kibana
kibana: 展示elasticsearch中的数据。

主机名	系统版本	安装软件	配置要求
pihao101	centos7.6	jdk1.8/Elasticsearch/zookeeper/kafka/kibana/nginx	4G
pihao102	centos7.6	jdk1.8/Elasticsearch/zookeeper/kafka/Logstash	4G
pihao103	centos7.6	jdk1.8/Elasticsearch/zookeeper/kafka/filebeat/nginx	4G

1、访问pihao103节点的nginx，生成本地日志

filebeat会采集日志并发送到kafka [test1] topic

filebeat kafka配置详解,kafka,elk,elasticsearch

2、查看kafka队列中的消息

[root@pihao101 kafka_2.11-2.0.0]# bin/kafka-console-consumer.sh --bootstrap-server pihao103:9092 --topic test1 --from-beginning

filebeat kafka配置详解,kafka,elk,elasticsearch

数据正常传输到了kafka中，logstash中没做数据过滤，只是转发了下，应该没问题。

3、使用之前安装的head插件查看elasticsearch的索引

访问pihao101:9100

filebeat kafka配置详解,kafka,elk,elasticsearch

4、kibana数据展示

访问pihao101,配置了登录验证

filebeat kafka配置详解,kafka,elk,elasticsearch

总结：本此操作只是记录下ELK的基本搭建流程，实际项目中采集的日志就是系统应用的日志咯，在配合搭建一个搜索的项目，方便开发人员分析日志,快速定位，后续的日志过滤场景之后再慢慢分享记录。

加油，少年！文章来源地址https://www.toymoban.com/news/detail-739509.html

到了这里，关于基于Filebeat、Kafka搭建ELK日志分析平台详细步骤的文章就介绍完了。如果您还想了解更多内容，请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章，希望大家以后多多支持TOY模板网！