ElasticStackFilebeatkafkaLogstashElasticsearch

Kafka部署

filebeat安装在要收集日志的应用服务器中&＃xff0c;filebeat收集到日志之后传输到kafka中&＃xff0c;logstash通过kafka拿到日志&＃xff0c;在由logstash传给后面的es&＃xff0c;es将日志传给后面的kibana&＃xff0c;最后通过kibana展示出来。

4、Kafka&＃xff1a;

​ 数据缓冲队列。同时提高了可扩展性。具有峰值处理能力&＃xff0c;使用消息队列能够使关键组件顶住突发的访问压力&＃xff0c;而不会因为突发的超负荷的请求而完全崩溃。是一个分布式、支持分区的&＃xff08;partition&＃xff09;、多副本的&＃xff08;replica&＃xff09;&＃xff0c;基于zookeeper协调的分布式消息系统&＃xff0c;它的最大的特性就是可以实时的处理大量数据以满足各种需求场景&＃xff1a;比如基于hadoop的批处理系统、低延迟的实时系统、web/nginx日志、访问日志&＃xff0c;消息服务等等&＃xff0c;用scala语言编写&＃xff0c;Linkedin于2010年贡献给了Apache基金会并成为顶级开源 项目。

Kafka的特性:

• 高吞吐量&＃xff1a;kafka每秒可以处理几十万条消息。

• 可扩展性&＃xff1a;kafka集群支持热扩展- 持久性、

• 可靠性&＃xff1a;消息被持久化到本地磁盘&＃xff0c;并且支持数据备份防止数据丢失

• 容错性&＃xff1a;允许集群中节点失败&＃xff08;若副本数量为n,则允许n-1个节点失败&＃xff09;

• 高并发&＃xff1a;支持数千个客户端同时读写

  它主要包括以下组件

  话题&＃xff08;Topic&＃xff09;&＃xff1a;是特定类型的消息流。(每条发布到 kafka 集群的消息属于的类别&＃xff0c;即 kafka 是面向 topic 的。)
生产者&＃xff08;Producer&＃xff09;&＃xff1a;是能够发布消息到话题的任何对象(发布消息到 kafka 集群的终端或服务).
消费者&＃xff08;Consumer&＃xff09;&＃xff1a;可以订阅一个或多个话题&＃xff0c;从而消费这些已发布的消息。
服务代理&＃xff08;Broker&＃xff09;&＃xff1a;已发布的消息保存在一组服务器中&＃xff0c;它们被称为代理&＃xff08;Broker&＃xff09;或Kafka集群。
partition&＃xff08;区&＃xff09;&＃xff1a;每个 topic 包含一个或多个 partition。
replication&＃xff1a;partition 的副本&＃xff0c;保障 partition 的高可用。
leader&＃xff1a;replica 中的一个角色&＃xff0c; producer 和 consumer 只跟 leader 交互。
follower&＃xff1a;replica 中的一个角色&＃xff0c;从 leader 中复制数据。
zookeeper&＃xff1a;kafka 通过 zookeeper 来存储集群的信息。


zookeeper:

ZooKeeper是一个分布式协调服务&＃xff0c;它的主要作用是为分布式系统提供一致性服务&＃xff0c;提供的功能包括&＃xff1a;配置维护、分布式同步等。Kafka的运行依赖ZooKeeper。 也是java微服务里面使用的一个注册中心服务

ZooKeeper主要用来协调Kafka的各个broker&＃xff0c;不仅可以实现broker的负载均衡&＃xff0c;而且当增加了broker或者某个broker故障了&＃xff0c;ZooKeeper将会通知生产者和消费者&＃xff0c;这样可以保证整个系统正常运转。

在Kafka中,一个topic会被分成多个区并被分到多个broker上&＃xff0c;分区的信息以及broker的分布情况与消费者当前消费的状态信息都会保存在ZooKeeper中&＃xff0c;

&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;

系统类型&＃xff1a;Centos7.5
节点IP&＃xff1a;192.168.246.234,192.168.246.231、192.168.246.235
软件版本&＃xff1a;jdk-8u121-linux-x64.tar.gz、kafka_2.11-2.1.0.tgz
示例节点&＃xff1a;172.16.246.231

1.安装配置jdk8

&＃xff08;1&＃xff09;Kafka、Zookeeper&＃xff08;简称&＃xff1a;ZK&＃xff09;运行依赖jdk8

tar zxvf /usr/local/package/jdk-8u121-linux-x64.tar.gz -C /usr/local/
echo &＃39;
JAVA_HOME&＃61;/usr/local/jdk1.8.0_121
PATH&＃61;$JAVA_HOME/bin:$PATH
export JAVA_HOME PATH
&＃39; >>/etc/profile
source /etc/profile


2.安装配置ZK

Kafka运行依赖ZK&＃xff0c;Kafka官网提供的tar包中&＃xff0c;已经包含了ZK&＃xff0c;这里不再额下载ZK程序。

配置相互解析—三台机器

[root&＃64;es-2-zk-log ~]# vim /etc/hosts
192.168.246.234 mes-1
192.168.246.231 es-2-zk-log
192.168.246.235 es-3-head-kib


&＃xff08;1&＃xff09;安装

[root&＃64;es-2-zk-log ~]# tar xzvf kafka_2.11-2.1.0.tgz -C /usr/local/

&＃xff08;2&＃xff09;配置

[root&＃64;mes-1 ~]# sed -i &＃39;s/^[^#]/#&/&＃39; /usr/local/kafka_2.11-2.1.0/config/zookeeper.properties
[root&＃64;mes-1 ~]# vim /usr/local/kafka_2.11-2.1.0/config/zookeeper.properties #添加如下配置
dataDir&＃61;/opt/data/zookeeper/data # 需要创建&＃xff0c;所有节点一致
dataLogDir&＃61;/opt/data/zookeeper/logs # 需要创建&＃xff0c;所有节点一致
clientPort&＃61;2181
tickTime&＃61;2000
initLimit&＃61;20
syncLimit&＃61;10 # 以下 IP 信息根据自己服务器的 IP 进行修改
server.1&＃61;192.168.246.231:2888:3888 //kafka集群IP:Port
server.2&＃61;192.168.246.234:2888:3888
server.3&＃61;192.168.246.235:2888:3888
#创建data、log目录
[root&＃64;mes-1 ~]# mkdir -p /opt/data/zookeeper/{data,logs}
#创建myid文件
[root&＃64;mes-1 ~]# echo 1 > /opt/data/zookeeper/data/myid #myid号按顺序排


[root&＃64;es-2-zk-log ~]# sed -i &＃39;s/^[^#]/#&/&＃39; /usr/local/kafka_2.11-2.1.0/config/zookeeper.properties
[root&＃64;es-2-zk-log ~]# vim /usr/local/kafka_2.11-2.1.0/config/zookeeper.properties
dataDir&＃61;/opt/data/zookeeper/data
dataLogDir&＃61;/opt/data/zookeeper/logs
clientPort&＃61;2181
tickTime&＃61;2000
initLimit&＃61;20
syncLimit&＃61;10
server.1&＃61;192.168.246.231:2888:3888
server.2&＃61;192.168.246.234:2888:3888
server.3&＃61;192.168.246.235:2888:3888
#创建data、log目录
[root&＃64;es-2-zk-log ~]# mkdir -p /opt/data/zookeeper/{data,logs}
#创建myid文件
[root&＃64;es-2-zk-log ~]# echo 2 > /opt/data/zookeeper/data/myid


[root&＃64;es-3 ~]# sed -i &＃39;s/^[^#]/#&/&＃39; /usr/local/kafka_2.11-2.1.0/config/zookeeper.properties
[root&＃64;es-3-head-kib ~]# vim /usr/local/kafka_2.11-2.1.0/config/zookeeper.properties
dataDir&＃61;/opt/data/zookeeper/data
dataLogDir&＃61;/opt/data/zookeeper/logs
clientPort&＃61;2181
tickTime&＃61;2000
initLimit&＃61;20
syncLimit&＃61;10
server.1&＃61;192.168.246.231:2888:3888
server.2&＃61;192.168.246.234:2888:3888
server.3&＃61;192.168.246.235:2888:3888
#创建data、log目录
[root&＃64;es-3-head-kib ~]# mkdir -p /opt/data/zookeeper/{data,logs}
#创建myid文件
[root&＃64;es-3-head-kib ~]# echo 3 > /opt/data/zookeeper/data/myid


配置项含义&＃xff1a;

dataDir ZK数据存放目录。
dataLogDir ZK日志存放目录。
clientPort 客户端连接ZK服务的端口。
tickTime ZK服务器之间或客户端与服务器之间维持心跳的时间间隔。
initLimit 允许follower连接并同步到Leader的初始化连接时间&＃xff0c;当初始化连接时间超过该值&＃xff0c;则表示连接失败。
syncLimit Leader与Follower之间发送消息时如果follower在设置时间内不能与leader通信&＃xff0c;那么此follower将会被丢弃。
server.1&＃61;172.16.244.31:2888:3888 2888是follower与leader交换信息的端口&＃xff0c;3888是当leader挂了时用来执行选举时服务器相互通信的端口。


3.配置Kafka

&＃xff08;1&＃xff09;配置

[root&＃64;mes-1 ~]# sed -i &＃39;s/^[^#]/#&/&＃39; /usr/local/kafka_2.11-2.1.0/config/server.properties
[root&＃64;mes-1 ~]# vim /usr/local/kafka_2.11-2.1.0/config/server.properties #在最后添加
broker.id&＃61;1
listeners&＃61;PLAINTEXT://192.168.246.231:9092
num.network.threads&＃61;3
num.io.threads&＃61;8
socket.send.buffer.bytes&＃61;102400
socket.receive.buffer.bytes&＃61;102400
socket.request.max.bytes&＃61;104857600
log.dirs&＃61;/opt/data/kafka/logs
num.partitions&＃61;6
num.recovery.threads.per.data.dir&＃61;1
offsets.topic.replication.factor&＃61;2
transaction.state.log.replication.factor&＃61;1
transaction.state.log.min.isr&＃61;1
log.retention.hours&＃61;168
log.segment.bytes&＃61;536870912
log.retention.check.interval.ms&＃61;300000
zookeeper.connect&＃61;192.168.246.231:2181,192.168.246.234:2181,192.168.246.235:2181
zookeeper.connection.timeout.ms&＃61;6000
group.initial.rebalance.delay.ms&＃61;0
[root&＃64;mes-1 ~]# mkdir -p /opt/data/kafka/logs


[root&＃64;es-2-zk-log ~]# sed -i &＃39;s/^[^#]/#&/&＃39; /usr/local/kafka_2.11-2.1.0/config/server.properties
[root&＃64;es-2-zk-log ~]# vim /usr/local/kafka_2.11-2.1.0/config/server.properties
broker.id&＃61;2
listeners&＃61;PLAINTEXT://192.168.246.234:9092
num.network.threads&＃61;3
num.io.threads&＃61;8
socket.send.buffer.bytes&＃61;102400
socket.receive.buffer.bytes&＃61;102400
socket.request.max.bytes&＃61;104857600
log.dirs&＃61;/opt/data/kafka/logs
num.partitions&＃61;6
num.recovery.threads.per.data.dir&＃61;1
offsets.topic.replication.factor&＃61;2
transaction.state.log.replication.factor&＃61;1
transaction.state.log.min.isr&＃61;1
log.retention.hours&＃61;168
log.segment.bytes&＃61;536870912
log.retention.check.interval.ms&＃61;300000
zookeeper.connect&＃61;192.168.246.231:2181,192.168.246.234:2181,192.168.246.235:2181
zookeeper.connection.timeout.ms&＃61;6000
group.initial.rebalance.delay.ms&＃61;0
[root&＃64;es-2-zk-log ~]# mkdir -p /opt/data/kafka/logs


[root&＃64;es-3-head-kib ~]# sed -i &＃39;s/^[^#]/#&/&＃39; /usr/local/kafka_2.11-2.1.0/config/server.properties
[root&＃64;es-3-head-kib ~]# vim /usr/local/kafka_2.11-2.1.0/config/server.properties
broker.id&＃61;3
listeners&＃61;PLAINTEXT://192.168.246.235:9092
num.network.threads&＃61;3
num.io.threads&＃61;8
socket.send.buffer.bytes&＃61;102400
socket.receive.buffer.bytes&＃61;102400
socket.request.max.bytes&＃61;104857600
log.dirs&＃61;/opt/data/kafka/logs
num.partitions&＃61;6
num.recovery.threads.per.data.dir&＃61;1
offsets.topic.replication.factor&＃61;2
transaction.state.log.replication.factor&＃61;1
transaction.state.log.min.isr&＃61;1
log.retention.hours&＃61;168
log.segment.bytes&＃61;536870912
log.retention.check.interval.ms&＃61;300000
zookeeper.connect&＃61;192.168.246.231:2181,192.168.246.234:2181,192.168.246.235:2181
zookeeper.connection.timeout.ms&＃61;6000
group.initial.rebalance.delay.ms&＃61;0
[root&＃64;es-3-head-kib ~]# mkdir -p /opt/data/kafka/logs


配置项含义&＃xff1a;

broker.id 每个server需要单独配置broker id&＃xff0c;如果不配置系统会自动配置。
listeners 监听地址&＃xff0c;格式PLAINTEXT://IP:端口。
num.network.threads
num.io.threads
socket.send.buffer.bytes
socket.receive.buffer.bytes
socket.request.max.bytes
log.dirs 日志文件目录。
num.partitions
num.recovery.threads.per.data.dir
offsets.topic.replication.factor
log.retention.hours
log.segment.bytes
log.retention.check.interval.ms
zookeeper.connect ZK主机地址&＃xff0c;如果zookeeper是集群则以逗号隔开。
zookeeper.connection.timeout.ms 连接到Zookeeper的超时时间。


4、其他节点配置

只需把配置好的安装包直接分发到其他节点&＃xff0c;修改 Kafka的broker.id和 listeners就可以了。

5、启动、验证ZK集群

&＃xff08;1&＃xff09;启动

在三个节点依次执行&＃xff1a;

[root&＃64;mes-1 ~]# cd /usr/local/kafka_2.11-2.1.0/
[root&＃64;mes-1 kafka_2.11-2.1.0]# nohup bin/zookeeper-server-start.sh config/zookeeper.properties &


&＃xff08;2&＃xff09;验证

查看端口

[root&＃64;mes-1 ~]# netstat -lntp | grep 2181
tcp6 0 0 :::2181 :::* LISTEN 1226/java

6、启动、验证Kafka

&＃xff08;1&＃xff09;启动

在三个节点依次执行&＃xff1a;

[root&＃64;mes-1 ~]# cd /usr/local/kafka_2.11-2.1.0/
[root&＃64;mes-1 kafka_2.11-2.1.0]# nohup bin/kafka-server-start.sh config/server.properties &


&＃xff08;2&＃xff09;验证

在192.168.246.231上创建topic

[root&＃64;es-2-zk-log kafka_2.11-2.1.0]# bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic testtopic
Created topic "testtopic".参数解释&＃xff1a;
–zookeeper指定zookeeper的地址和端口&＃xff0c;
–partitions指定partition的数量&＃xff0c;
–replication-factor指定数据副本的数量


在246.235上面查询192.168.246.231上的topic

[root&＃64;es-3-head-kib kafka_2.11-2.1.0]# bin/kafka-topics.sh --zookeeper 192.168.246.231:2181 --list
testtopic


模拟消息生产和消费
发送消息到192.168.246.231

[root&＃64;mes-1 kafka_2.11-2.1.0]# bin/kafka-console-producer.sh --broker-list 192.168.122.6:9092 --topic testtopic
>hello

从192.168.246.234接受消息

[root&＃64;es-2-zk-log kafka_2.11-2.1.0]# bin/kafka-console-consumer.sh --bootstrap-server 192.168.122.106:9092 --topic testtopic --from-beginning
hello

kafka没有问题之后&＃xff0c;回到logstach服务器&＃xff1a;
#安装完kafka之后的操作&＃xff1a;
[root&＃64;es-2-zk-log ~]# cd /usr/local/logstash-6.5.4/etc/conf.d/
[root&＃64;es-2-zk-log conf.d]# cp input.conf input.conf.bak
[root&＃64;es-2-zk-log conf.d]# vim input.conf
input {
kafka { #指定kafka服务type &＃61;> "nginx_log"codec &＃61;> "json" #通用选项&＃xff0c;用于输入数据的编解码器topics &＃61;> "nginx" #这里定义的topicdecorate_events &＃61;> truebootstrap_servers &＃61;> "192.168.246.234:9092, 192.168.246.231:9092, 192.168.246.235:9092"}
}
启动 logstash
[root&＃64;es-2-zk-log conf.d]# cd /usr/local/logstash-6.5.4/
[root&＃64;es-2-zk-log logstash-6.5.4]# nohup bin/logstash -f etc/conf.d/ --config.reload.automatic &


decorate_events &＃61;> true 默认是 false&＃96; 这将向logstash 事件添加一个名为kafka的字段 &＃xff0c;这包含以下属性。

• &＃96;topic 主题&＃xff1a;与此消息相关联的主题

• consumer_group 使用者群组&＃xff1a;此事件中用来读取的使用者群组

• partition 分区&＃xff1a;与此消息关联的分区

• offset 偏移量&＃xff1a;与此消息关联的分区的偏移量

• key&＃xff1a;包含message key的ByteBuffer

二、配置 Filebeat 输出到 kafka

1 配置

output.kafka:# initial brokers for reading cluster metadatahosts: ["kafka1:9092", "kafka2:9092", "kafka3:9092"]topic: &＃39;nginx&＃39;partition.round_robin:reachable_only: falserequired_acks: 1compression: gzipmax_message_bytes: 1000000


2 验证

验证kafka是否生成topic

[root&＃64;es-3-head-kib filebeat]# cd /usr/local/kafka_2.11-2.1.0/
[root&＃64;es-3-head-kib kafka_2.11-2.1.0]# bin/kafka-topics.sh --zookeeper 192.168.246.231:2181 --list
__consumer_offsets
nginx #已经生成topic
testtopic


三、配置 Logstash 从 kafka 中获取日志

input {kafka {bootstrap_servers &＃61;> "myhost:9092"topics &＃61;> ["nginx"]codec &＃61;> json}
}