devbox
AllinToyou
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

7.kafka+ELK连接_elk接入kafka

作者：AllinToyou | 2024-04-24 08:00:52

赞

踩

elk接入kafka

文章目录

kafka+ELK连接

部署Kafka

###关闭防火墙

systemctl stop firewalld
systemctl disable firewalld

setenforce 0

vim /etc/selinux/config

SELINUX=disabled

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
###下载安装包

官方下载地址：http://kafka.apache.org/downloads.html

cd /opt
wget https://mirrors.tuna.tsinghua.edu.cn/apache/kafka/2.7.1/kafka_2.13-2.7.1.tgz

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
###安装 Kafka

cd /opt/
tar xf kafka_2.13-2.7.1.tgz
mv kafka_2.13-2.7.1 /usr/local/kafka

  • 1
  • 2
  • 3
  • 4
  • 5
###修改配置文件

##备份配置文件
cd /usr/local/kafka/config/
cp server.properties{,.bak}

vim server.properties

---21行--
broker.id=0    
###21行，broker的全局唯一编号，每个broker不能重复，因此要在其他机器上配置 broker.id=1、broker.id=2

---31行---
listeners=PLAINTEXT://192.168.242.70:9092    
###31行，指定监听的IP和端口，如果修改每个broker的IP需区分开来，也可保持默认配置不用修改

---42行---
num.network.threads=3    
###42行，broker 处理网络请求的线程数量，一般情况下不需要去修改

---45行---
num.io.threads=8            
#45行，用来处理磁盘IO的线程数量，数值应该大于硬盘数

---48行---
socket.send.buffer.bytes=102400       
#48行，发送套接字的缓冲区大小

---51行---
socket.receive.buffer.bytes=102400    
#51行，接收套接字的缓冲区大小

---54行---
socket.request.max.bytes=104857600    
#54行，请求套接字的缓冲区大小

---60行---
log.dirs=/usr/local/kafka/logs        
#60行，kafka运行日志存放的路径，也是数据存放的路径

---65行---
num.partitions=1    
#65行，topic在当前broker上的默认分区个数，会被topic创建时的指定参数覆盖

---69行---
num.recovery.threads.per.data.dir=1    
#69行，用来恢复和清理data下数据的线程数量

---103行---
log.retention.hours=168    
#103行，segment文件（数据文件）保留的最长时间，单位为小时，默认为7天，超时将被删除

---110行---
log.segment.bytes=1073741824    
#110行，一个segment文件最大的大小，默认为 1G，超出将新建一个新的segment文件

---123行---
zookeeper.connect=192.168.242.70:2181,192.168.242.71:2181,192.168.242.72:2181    
###123行，配置连接Zookeeper集群地址

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35
  • 36
  • 37
  • 38
  • 39
  • 40
  • 41
  • 42
  • 43
  • 44
  • 45
  • 46
  • 47
  • 48
  • 49
  • 50
  • 51
  • 52
  • 53
  • 54
  • 55
  • 56
  • 57
  • 58
  • 59
###修改环境变量

vim /etc/profile
export KAFKA_HOME=/usr/local/kafka
export PATH=$PATH:$KAFKA_HOME/bin

###加载配置
source /etc/profile


###复制文件到zookeeper服务器哦

scp -r /usr/local/kafka 192.168.242.71:/usr/local
scp -r /usr/local/kafka 192.168.242.72:/usr/local

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
###部署kafka启动脚本

vim /etc/init.d/kafka


#!/bin/bash
#chkconfig:2345 22 88
#description:Kafka Service Control Script
KAFKA_HOME='/usr/local/kafka'
case $1 in
start)
	echo "---------- Kafka 启动 ------------"
	${KAFKA_HOME}/bin/kafka-server-start.sh -daemon ${KAFKA_HOME}/config/server.properties
;;
stop)
	echo "---------- Kafka 停止 ------------"
	${KAFKA_HOME}/bin/kafka-server-stop.sh
;;
restart)
	$0 stop
	$0 start
;;
status)
	echo "---------- Kafka 状态 ------------"
	count=$(ps -ef | grep kafka | egrep -cv "grep|$$")
	if [ "$count" -eq 0 ];then
        echo "kafka is not running"
    else
        echo "kafka is running"
    fi
;;
*)
    echo "Usage: $0 {start|stop|restart|status}"
esac

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
###设置开机自启

cd /etc/init.d/
chmod +x /etc/init.d/kafka
chkconfig --add kafka

###分别启动 Kafka
service kafka start


###另外一种启动方式

cd /usr/local/kafka/bin

./kafka-server-start.sh -daemon /usr/local/kafka/config/server.properties

netstat -lntp | grep 9092

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17

在这里插入图片描述

kafka操作命令

####创建topic

kafka-topics.sh --create --zookeeper 192.168.242.70:2181,192.168.242.71:2181,192.168.242.72:2181 --replication-factor 2 --partitions 3 --topic test



--zookeeper：定义 zookeeper 集群服务器地址，如果有多个 IP 地址使用逗号分割，一般使用一个 IP 即可
--replication-factor：定义分区副本数，1 代表单副本，建议为 2 
--partitions：定义分区数 
--topic：定义 topic 名称

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10

在这里插入图片描述

###查看当前服务器中的所有 topic

kafka-topics.sh --list --zookeeper 192.168.242.70:2181,192.168.242.71:2181,192.168.242.72:2181 

  • 1
  • 2
  • 3
###查看某个 topic 的详情

kafka-topics.sh  --describe --zookeeper 192.168.242.70:2181,192.168.242.71:2181,192.168.242.72:2181 

  • 1
  • 2
  • 3

在这里插入图片描述

###发布消息

kafka-console-producer.sh --broker-list 192.168.242.70:9092,192.168.242.71:9092,192.168.242.72:9092  --topic test

  • 1
  • 2
  • 3
###消费消息

kafka-console-consumer.sh --bootstrap-server 192.168.242.70:9092,192.168.242.71:9092,192.168.242.72:9092 --topic test --from-beginning

--from-beginning：会把主题中以往所有的数据都读取出来

  • 1
  • 2
  • 3
  • 4
  • 5

在这里插入图片描述

###修改分区数

kafka-topics.sh --zookeeper 192.168.242.70:2181,192.168.242.71:2181,192.168.242.72:2181  --alter --topic test --partitions 6

  • 1
  • 2
  • 3

在这里插入图片描述

###删除 topic

kafka-topics.sh --delete --zookeeper 192.168.242.70:2181,192.168.242.71:2181,192.168.242.72:2181 --topic test

  • 1
  • 2
  • 3

在这里插入图片描述

kafka架构深入

  • Kafka 工作流程及文件存储机制

    • Kafka 中消息是以 topic 进行分类的，生产者生产消息，消费者消费消息，都是面向 topic 的。

  • topic 是逻辑上的概念，而 partition 是物理上的概念，每个 partition 对应于一个 log 文件，该 log 文件中存储的就是 producer 生产的数据。

  • Producer 生产的数据会被不断追加到该 log 文件末端，且每条数据都有自己的 offset。

  • 消费者组中的每个消费者，都会实时记录自己消费到了哪个 offset，以便出错恢复时，从上次的位置继续消费。

  • 由于生产者生产的消息会不断追加到 log 文件末尾，为防止 log 文件过大导致数据定位效率低下，Kafka 采取了分片和索引机制，将每个 partition 分为多个 segment。

  • 每个 segment 对应两个文件：“.index” 文件和 “.log” 文件。

  • 这些文件位于一个文件夹下，该文件夹的命名规则为：topic名称+分区序号。例如，test 这个 topic 有三个分区， 则其对应的文件夹为 test-0、test-1、test-2。

  • index 和 log 文件以当前 segment 的第一条消息的 offset 命名。

    • “.index” 文件存储大量的索引信息，
    • “.log” 文件存储大量的数据，
    • 索引文件中的元数据指向对应数据文件中 message 的物理偏移地址。

  • 数据可靠性保证

    • 为保证 producer 发送的数据，能可靠的发送到指定的 topic，topic 的每个 partition 收到 producer 发送的数据后， 都需要向 producer 发送 ack（acknowledgement 确认收到），如果 producer 收到 ack，就会进行下一轮的发送，否则重新发送数据。

  • 数据一致性问题

    • LEO：指的是每个副本最大的 offset；
    • HW：指的是消费者能见到的最大的 offset，所有副本中最小的 LEO。

  • follower 故障

    • follower 发生故障后会被临时踢出 ISR（Leader 维护的一个和 Leader 保持同步的 Follower 集合），待该 follower 恢复后，follower 会读取本地磁盘记录的上次的 HW，并将 log 文件高于 HW 的部分截取掉，从 HW 开始向 leader 进行同步。等该 follower 的 LEO 大于等于该 Partition 的 HW，即 follower 追上 leader 之后，就可以重新加入 ISR 了。

  • leader 故障

    • leader 发生故障之后，会从 ISR 中选出一个新的 leader， 之后，为保证多个副本之间的数据一致性，其余的 follower 会先将各自的 log 文件高于 HW 的部分截掉，然后从新的 leader 同步数据。

  • 注：这只能保证副本之间的数据一致性，并不能保证数据不丢失或者不重复。

  • ack 应答机制

    • 对于某些不太重要的数据，对数据的可靠性要求不是很高，能够容忍数据的少量丢失，所以没必要等 ISR 中的 follower 全部接收成功。
    • 所以 Kafka 为用户提供了三种可靠性级别，用户根据对可靠性和延迟的要求进行权衡选择。

  • 当 producer 向 leader 发送数据时，可以通过 request.required.acks 参数来设置数据可靠性的级别：

    • 0：这意味着producer无需等待来自broker的确认而继续发送下一批消息。这种情况下数据传输效率最高，但是数据可靠性确是最低的。当broker故障时有可能丢失数据。
    • 1（默认配置）：这意味着producer在ISR中的leader已成功收到的数据并得到确认后发送下一条message。如果在follower同步成功之前leader故障，那么将会丢失数据。
    • -1（或者是all）：producer需要等待ISR中的所有follower都确认接收到数据后才算一次发送完成，可靠性最高。但是如果在 follower 同步完成后，broker 发送ack 之前，leader 发生故障，那么会造成数据重复。

  • 三种机制性能依次递减，数据可靠性依次递增。

注：在 0.11 版本以前的Kafka，对此是无能为力的，只能保证数据不丢失，再在下游消费者对数据做全局去重。在 0.11 及以后版本的 Kafka，引入了一项重大特性：幂等性。所谓的幂等性就是指 Producer 不论向 Server 发送多少次重复数据， Server 端都只会持久化一条。

+ kafka会通过ack机制保证数据的可靠性

  • ack配置参数有
  • 0（效果类似异步复制）：不等待follower同步完成就让生产者发送下一条消息
  • 1（效果类似半同步复制）：至少等待一个follower同步完成才让生产者发送下一条消息
  • -1（效果类似全同步复制）：等待所有follower同步完成才让生产者发送下一条消息

Filebeat+Kafka+ELK连接

在这里插入图片描述

###部署 Zookeeper+Kafka 集群

###部署 Filebeat，修改配置文件 

cd /usr/local/filebeat

vim filebeat.yml


filebeat.prospectors:
- type: log
  enabled: true
  paths:
    - /var/log/httpd/access_log
  tags: ["access"]


- type: log
  enabled: true
  paths:
    - /var/log/httpd/error_log
  tags: ["error"]
  
......


#添加输出到 Kafka 的配置

output.kafka:
  enabled: true
  hosts: ["192.168.242.70:9092","192.168.242.71:9092","192.168.242.72:9092"]    #指定 Kafka 集群配置
  topic: "httpd"    #指定 Kafka 的 topic
  

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33

在这里插入图片描述
在这里插入图片描述

###启动 filebeat

./filebeat -e -c filebeat.yml

  • 1
  • 2
  • 3

在这里插入图片描述

###部署 ELK，在 Logstash 组件所在节点上新建一个 Logstash 配置文件

cd /etc/logstash/conf.d/

vim kafka.conf

input {
    kafka {
        bootstrap_servers => "192.168.242.70:9092,192.168.242.71:9092,192.168.242.72:9092"  
    #kafka集群地址
    
        topics  => "httpd"     #拉取的kafka的指定topic
        type => "httpd_kafka"  #指定 type 字段
        codec => "json"        #解析json格式的日志数据
        auto_offset_reset => "latest"  #拉取最近数据，earliest为从头开始拉取
        decorate_events => true   #传递给elasticsearch的数据额外增加kafka的属性数据
    }
}

output {
  if "access" in [tags] {
    elasticsearch {
      hosts => ["192.168.242.66:9200"]
      index => "httpd_access-%{+YYYY.MM.dd}"
    }
  }
  
  if "error" in [tags] {
    elasticsearch {
      hosts => ["192.168.242.66:9200"]
      index => "httpd_error-%{+YYYY.MM.dd}"
    }
  }
  
  stdout { codec => rubydebug }
}


  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35
  • 36
  • 37
####启动 logstash

logstash -f kafka.conf

  • 1
  • 2
  • 3

在这里插入图片描述
在这里插入图片描述

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/AllinToyou/article/detail/478373
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号