166. ELK+filebeat+kafka+zookeeper构建海量日志分析平台
程序员文章站
2022-03-27 20:43:58
...
1. 消息队列基本介绍
1. 什么是消息队列?
消息Message :比如两个设备进行数据的传输,所传输的数据,都可以称为消息,可以是文
本、音频
队列Queue :是一种”先进先出 的数据结构。类似排队买票、羽毛球筒。
消息队列MQ :是用来保存消息的一个容器。 消息队列需要两个功能接[ 7供外部调用,一个
是生产一个是消费。主要是进行数据存储和读取。把数据放到消息队列叫做生产者,从队列里
取数据叫做消费者。
2. MQ.主要分为两类点对点发布/订阅
共同点:
消息的生产者(Producer)生产消息发送到队列中,然后消息的消费者(Consumer)从队列中读取并消费消息。
不同点:
点对点:消息队列(Queue)、发送者(Sender)、接收者(Receiver)
一个生产者生产的消息只能有一个消费者,消息一旦被消费,消息就不在消息队列中了,比如:
钉钉的澡堂模式、打电话等。都是消息发送到消息队列后只能被一个接收者接收, 当接收完毕消息则销毁。
发布/订阅:消息队列(Queue)、发布者(Publisher)、 订阅者(Subscriber)、 主题( Topic)
每个消息可以有多个消费者,彼此互不影响,比如:我使用公众号发布一篇文章,关注我的人都
能看到,即发布到消息队列的消息能被多个接收者(订阅者)接收。
2. 消息队列使用场景
- 1.解耦
- 2.异步
- 3.削峰
3. kafka介绍
3.1 什么是kafka
kafka是一个实时数据处理系统。 实时数据处理系统就是数据一旦产生 ,I就要能快速进行处理的系统。对于实时数据处理系统,最常见的就是消息队列。kafka 也是一1MQ消息队列。
3.2 kafka的特点
高吞吐量:可以满足每秒百万级别消息的生产和消费。
持久性:有一套完善的消息存储机制,确保数据的高效安全的持久化。
分布式:基于分布式的扩展和容错机制,当某一台发生故障失效时 ,可以实现故障自动转移。
3.3 Kafka架构
Broker : kafka集群中包含多个kafka服务掉,每一个kafka服务节点就称为一个broker
Topic:主题, Kafka将消息分J别类,每-类的消息称之为( Topic).(Kafka消息数据是存储在硬盘上的
Partition:分区,每个Topic包含一个或多 个Partition ,在创建Topic时指定包含的Partition数量(目的是为了进行分布式存储
Replication:副本,每个分区可以有多副本,分布在不同的Broker上,选出一个副本作为Leader ,所有的读写
请求都会通过Leader完成,但Follower只负责备份数据,所有Follower会自动的从Leader中复制数据,当Leader
宕机后,会从Follower中选出一个 新的Leader继续提供服务,实现故障自动转移。
Message:消息,是通信的基本单位,每个消息都属于-个Partition
Producer:消息的生产者,向Kafka的一个topic发布消息
Consumer:消息的消费者,订阅topic并读取其发布的消息
Consumer Group:每个Consumer属于-个特定的Consumer Group ,多个Consumer可以属于同一个ConsumerGroup中
Zookeeper:要用来存储Kafka的元数据信息 ,比如:有多少集群节点主题名称、主要用来协调kafka的正常运行。但发送给Topic本身的消息数据并不存储在ZK中,而存在kafka的磁盘文件中。
4. kafka基本使用
1. 安装zookeeper
[root@kafka--node1 ~]# yum install java maven -y
[root@kafka-node1 ~]# tar xf apache-zookeeper-3.5.6-bin.tar.gz
[root@kafka-node1 ~]# cd apache-zookeeper-3.5.6-bin/conf/
[root@kafka-node1 conf]# cp zoo_sample.cfg zoo.cfg
[root@kafka-node1 conf]# grep -v "^#" zoo.cfg
tickTime=2000
initLimit=10
syncLimit=5
dataDir=../data
clientPort=2181
[root@kafka-node1 conf]# mkdir ../data
[root@kafka-node1 conf]# cd ../bin/
[root@kafka-node1 bin]# ./zkServer.sh start
[root@kafka-node1 bin]# ./zkServer.sh status
/usr/bin/java
ZooKeeper JMX enabled by default
Using config: /opt/apache-zookeeper-3.5.6-bin/bin/../conf/zoo.cfg
Client port found: 2181. Client address: localhost.
Mode: follower
2. 连接测试:
[root@kafka-node1 bin]# ./zkCli.sh
WatchedEvent state:SyncConnected type:None path:null
[zk: localhost:2181(CONNECTED) 0]
[zk: localhost:2181(CONNECTED) 0]
[zk: localhost:2181(CONNECTED) 0]
3. 安装kafka-
[root@kafka-node1 ~]# tar xf kafka-_2.12-2.3.0.tgz
[root@kafka-node1 ~]# cd /opt/kafka-_2.12-2.3.0/bin/
[root@kafka-node1 bin]# ./kafka--server-start.sh ../config/server.properties
检测9092(kafka-)端口是否开启
检测2181(zookeeper)端口是否开启
[root@kafka-node1 ~]# cd /opt/apache-zookeeper-3.5.6-bin/bin/
[root@kafka-node1 bin]# ./zkCli.sh
[zk: localhost:2181(CONNECTED) 0] ls /
[admin, brokers, cluster, config, consumers, controller, controller_epoch, isr_change_notification, latest_producer_id_block, log_dir_event_notification, zookeeper]
[zk: localhost:2181(CONNECTED) 2] ls /brokers
[ids, seqid, topics]
[root@kafka-node1 bin]# jps
6436 kafka-
1066 cerebro.cerebro-0.8.5-launcher.jar
11739 Jps
3071 QuorumPeerMain
4. 使用kafka-创建topics
[root@kafka-node1 bin]# cd /opt/kafka-_2.12-2.3.0/bin/
[root@kafka-node1 bin]# ./kafka--topics.sh \
> --create \
> --zookeeper localhost:2181 \
> --replication-factor 1 \
> --partitions 3 \
> --topic cheng
OpenJDK 64-Bit Server VM warning: If the number of processors is expected to increase from one, then you should configure the number of parallel GC threads appropriately using -XX:ParallelGCThreads=N
Created topic cheng.
5. 查看topic详情
[root@kafka-node1 bin]# cd /opt/apache-zookeeper-3.5.6-bin/bin/
[root@kafka-node1 bin]# ./zkCli.sh
[zk: localhost:2181(CONNECTED) 1] ls /brokers/topics/cheng/partitions
[0, 1, 2]
6. producer模拟生产者,产生数据
[root@kafka-node1 bin]# ./kafka--console-producer.sh --broker-list localhost:9092 --topic cheng
7. consumer模拟消费者,模拟消费
[root@kafka-node1 bin]# ./kafka--console-consumer.sh \
--bootstrap-server localhost:9092 \
--topic cheng \
--from-beginning
8. 删除topic
[root@kafka-node1 bin]# ./kafka--topic.sh \
--delete \
zookeeper localhost:2181 \
--topic cheng \
5. kafka集群部署
1.安装zookeeper集群
10.0.0.161
10.0.0.162
10.0.0.163
https://www.jianshu.com/p/1a99eebdf075
1.安装java
yum install -y java maven
2.配置zookeeper
[root@kafka-node1 conf]# cat zoo.cfg
# 服务器之间或客户端与服务器之间维持心跳的时间间隔
# tickTime以毫秒为单位。
tickTime=2000
# 集群中的follower服务器(F)与leader服务器(L)之间的初始连接心跳数
initLimit=10
# 集群中的follower服务器与leader服务器之间请求和应答之间能容忍的最多心跳数
syncLimit=5
# 数据保存目录
dataDir=../data
# 日志保存目录
dataLogDir=../logs
# 客户端连接端口
clientPort=2181
# 客户端最大连接数。# 根据自己实际情况设置,默认为60个
maxClientCnxns=60
# 三个接点配置,格式为: server.服务编号=服务地址、LF通信端口、选举端口
server.1=10.0.0.161:2888:3888
server.2=10.0.0.162:2888:3888
server.3=10.0.0.163:2888:3888
3.创建数据存储目录
[root@kafka-node1 conf]# mkdir ../data
4.将配置好的zookeeper拷贝至另外两个节点
[root@kafka-node1 opt]# scp -rp apache-zookeeper-3.5.6-bin aaa@qq.com:/opt
[root@kafka-node1 opt]# scp -rp apache-zookeeper-3.5.6-bin aaa@qq.com:/opt
5.在三个节点上,分别在/opt/apache-zookeeper-3.5.6-bin/data/myid写入节点标记:
node1的操作
[root@kafka-node1 opt]# echo "1" > /opt/apache-zookeeper-3.5.6-bin/data/myid
node2的操作
[root@kafka-node2 ~]# echo "2" > /opt/apache-zookeeper-3.5.6-bin/data/myid
node3的操作
[root@kafka-node3 ~]# echo "3" > /opt/apache-zookeeper-3.5.6-bin/data/myid
6.启动zookeeper集群
[root@kafka-node1 ~]# cd /opt/apache-zookeeper-3.5.6-bin/bin/
[root@kafka-node1 bin]# ./zkServer.sh start
7.检查集群状态
[root@kafka-node3 bin]# ./zkServer.sh status
/usr/bin/java
ZooKeeper JMX enabled by default
Using config: /opt/apache-zookeeper-3.5.6-bin/bin/../conf/zoo.cfg
Client port found: 2181. Client address: localhost.
Mode: leader
8. kafka集群安装
0.初始化环境
[root@kafka-node1 logs]# cat /etc/hosts
10.0.0.161 kafka-node1
10.0.0.162 kafka-node2
10.0.0.163 kafka-node3
1.解压kafka压缩包
[root@kafka-node1 ~]# cp kafka_2.12-2.3.0.tgz /opt/
[root@kafka-node1 ~]# cd /opt/
[root@kafka-node1 opt]# tar xf kafka_2.12-2.3.0.tgz
2.配置kafka
[root@kafka-node1 config]# cat server.properties
############################# Server Basics #############################
# broker的id,值为整数,且必须唯一,在一个集群中不能重复
broker.id=1
############################# Socket Server Settings #############################
# kafka默认监听的端口为9092 (默认与主机名进行连接)
listeners=PLAINTEXT://10.0.0.161:9092
# 处理网络请求的线程数量,默认为3个
num.network.threads=3
# 执行磁盘IO操作的线程数量,默认为8个
num.io.threads=8
# socket服务发送数据的缓冲区大小,默认100KB
socket.send.buffer.bytes=102400
# socket服务接受数据的缓冲区大小,默认100KB
socket.receive.buffer.bytes=102400
# socket服务所能接受的一个请求的最大大小,默认为100M
socket.request.max.bytes=104857600
############################# Log Basics #############################
# kafka存储消息数据的目录
log.dirs=../data
# 每个topic默认的partition数量
num.partitions=3
# 在启动时恢复数据和关闭时刷新数据时每个数据目录的线程数量
num.recovery.threads.per.data.dir=1
############################# Log Flush Policy #############################
# 消息刷新到磁盘中的消息条数阈值
#log.flush.interval.messages=10000
# 消息刷新到磁盘中的最大时间间隔,1s
#log.flush.interval.ms=1000
############################# Log Retention Policy #############################
# 日志保留小时数,超时会自动删除,默认为7天
log.retention.hours=168
# 日志保留大小,超出大小会自动删除,默认为1G
#log.retention.bytes=1073741824
# 日志分片策略,单个日志文件的大小最大为1G,超出后则创建一个新的日志文件
log.segment.bytes=1073741824
# 每隔多长时间检测数据是否达到删除条件,300s
log.retention.check.interval.ms=300000
############################# Zookeeper #############################
# Zookeeper连接信息,如果是zookeeper集群,则以逗号隔开
zookeeper.connect=10.0.0.161:2181,10.0.0.162:2181,10.0.0.163:2181
# 连接zookeeper的超时时间,6s
zookeeper.connection.timeout.ms=6000
3.创建数据存储的目录
[root@kafka-node1 config]# mkdir ../data
[root@kafka-node1 opt]# scp -rp kafka_2.12-2.3.0 aaa@qq.com:/opt
[root@kafka-node1 opt]# scp -rp kafka_2.12-2.3.0 aaa@qq.com:/opt
4.修改162和163的server.properties文件中的broker.id
node1
broker.id=1
node2
broker.id=2
node3
broker.id=3
5.启动kafka集群
[root@kafka-node2 bin]# export KAFKA_HEAP_OPTS="-Xmx256M -Xms128M"
[root@kafka-node2 bin]# ./kafka-server-start.sh ../config/server.properties #启动测试
[root@kafka-node2 bin]# ./kafka-server-start.sh -daemon ../config/server.properties #放入后台
6.使用kafka创建一个topic
[root@yinwu-kafka-node1-172 bin]# ./kafka-topics.sh \
--create \
--zookeeper 10.0.0.161:2181,10.0.0.162:2181,10.0.0.163:2181 \
--partitions 3 \
--replication-factor 1 \
--topic yinwu-kafka
7.消息发布者测试
[root@yinwu-kafka-node1-172 bin]# ./kafka-console-producer.sh \
--broker-list 10.0.0.161:9092,10.0.0.162:9092,10.0.0.163:9092 \
--topic yinwu-kafka
8.消息订阅者测试
[root@yinwu-kafka-node1-172 bin]# ./kafka-console-consumer.sh \
--bootstrap-server 10.0.0.161:9092,10.0.0.162:9092,10.0.0.163:9092 \
--topic yinwu-kafka \
--from-beginning
6. ELK对接kafka
[root@web01 filebeat]# cat filebeat.yml
filebeat.inputs:
- type: log
enabled: true
paths:
- /var/log/nginx/access.log
tags: ["access"]
- type: log
enabled: true
paths:
- /var/log/nginx/error.log
tags: ["error"]
output.kafka:
hosts: ["10.0.0.161:9092", "10.0.0.162:9092", "10.0.0.163:9092"] #kafka集群地址
topics:
- topic: "kafka-nginx-access" #topic名称
when.contains:
tags: "access" #当时tags是acceess时,则写入kafka-nginx-access topic中
- topic: "kafka-nginx-error" #topic名称
when.contains:
tags: "error" #当时tags是error时,则写入kafka-nginx-error topic中
[root@logstash-node1 conf.d]# cat input_filebeat_kafka_output_es.conf
input {
kafka {
bootstrap_servers => "10.0.0.161:9092,10.0.0.162:9092,10.0.0.163:9092"
topics => ["kafka-nginx-access"]
group_id => "logstash" #消费组名称
consumer_threads => 2 #消费者启动的线程,尽量与分区一致
codec => "json" #指定格式为json
}
kafka {
bootstrap_servers => "10.0.0.161:9092,10.0.0.162:9092,10.0.0.163:9092"
topics => ["kafka-nginx-error"]
group_id => "logstash" #消费组名称
consumer_threads => 2 #消费者启动的线程,尽量与分区一致
codec => "json" #指定格式为json
}
}
filter {
if "access" in [tags][0] {
grok {
match => { "message" => "%{IPORHOST:clientip} %{USER:ident} %{USER:auth} \[%{HTTPDATE:timestamp}\] \"%{WORD:method} %{URIPATHPARAM:request} HTTP/%{NUMBER:httpversion}\" %{NUMBER:response} (?:%{NUMBER:bytes}|-) %{QS:hostname} (?:%{QS:referrer}|-) (?:%{NOTSPACE:post_args}|-) %{QS:useragent} (?:%{QS:x_forward_for}|-) (?:%{URIHOST:upstream_host}|-) (?:%{NUMBER:upstream_response_code}|-) (?:%{NUMBER:upstream_response_time}|-) (?:%{NUMBER:response_time}|-)" }
}
useragent {
source => "useragent"
target => "useragent"
}
geoip {
source => "clientip"
}
date {
match => ["timestamp","dd/MMM/yyyy:HH:mm:ss Z"]
target => "@timestamp"
timezone => "Asia/Shanghai"
}
mutate {
convert => ["bytes","integer"]
convert => ["response_time", "float"]
remove_field => ["message","agent","tags"]
add_field => { "target_index" => "logstash-kafka-nginx-access-%{+YYYY.MM}" }
}
#提取referrer具体的域名/^"http/
if [referrer] =~ /^"http/ {
grok {
match => { "referrer" => '%{URIPROTO}://%{URIHOST:referrer_host}' }
}
}
#提取用户请求资源类型以及资源ID编号
if "oldxu.com" in [referrer_host] {
grok {
match => { "referrer" => '%{URIPROTO}://%{URIHOST}/(%{NOTSPACE:oldxu_type}/%{NOTSPACE:oldxu_res_id})?"' }
}
}
}
else if "error" in [tags][0] {
date {
match => ["timestamp","dd/MMM/yyyy:HH:mm:ss Z"]
target => "@timestamp"
timezone => "Asia/Shanghai"
}
mutate {
add_field => { "target_index" => "logstash-kafka-nginx-error-%{+YYYY.MM}" }
}
}
}
output {
stdout {
codec => rubydebug
}
elasticsearch {
hosts => ["10.0.0.161:9200","10.0.0.162:9200","10.0.0.163:9200"]
index => "%{[target_index]}"
template_overwrite => true
}
}
上一篇: 全栈python第十六天
下一篇: 5 个免费的 Web 视频及音频播放器