淘先锋技术网

首页 1 2 3 4 5 6 7

kafka+ELK连接

部署Kafka

###关闭防火墙

systemctl stop firewalld
systemctl disable firewalld

setenforce 0

vim /etc/selinux/config

SELINUX=disabled
###下载安装包

官方下载地址: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
###安装 Kafka

cd /opt/
tar xf kafka_2.13-2.7.1.tgz
mv kafka_2.13-2.7.1 /usr/local/kafka
###修改配置文件

##备份配置文件
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集群地址
###修改环境变量

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
###部署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
###设置开机自启

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

在这里插入图片描述

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 名称

在这里插入图片描述

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

kafka-topics.sh --list --zookeeper 192.168.242.70:2181,192.168.242.71:2181,192.168.242.72:2181 
###查看某个 topic 的详情

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

在这里插入图片描述

###发布消息

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

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:会把主题中以往所有的数据都读取出来

在这里插入图片描述

###修改分区数

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

在这里插入图片描述

###删除 topic

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

在这里插入图片描述

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
  

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

###启动 filebeat

./filebeat -e -c filebeat.yml

在这里插入图片描述

###部署 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 }
}

####启动 logstash

logstash -f kafka.conf

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