kafka常用脚本_kafka脚本

kafka相关的脚本

kafka脚本

kafka启动，可写成如下脚本

## 启动：kafka.sh start

#! /bin/bash
case $1 in
"start"){
 for i in node04 node05 node06
 do
 echo " --------启动 $i Kafka-------"
 ssh $i "source /etc/profile ; /opt/software/kafka/bin/kafka-server-start.sh -daemon /opt/software/kafka/config/server.properties"
 done
};;
"stop"){
 for i in node04 node05 node06
 do
 echo " --------停止 $i Kafka-------"
 ssh $i "source /etc/profile; /opt/software/kafka/bin/kafka-server-stop.sh  "
 done
};;
esac
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

列举topic：	
/bin/kafka-topics.sh --bootstrap-server node05:9092 --list
dsp_bid
dsp_click
dsp_monitor 

###新建topic 
/bin/kafka-topics.sh --bootstrap-server node05:9092 --create  --replication-factor 2 --partitions 3 --topic t_monitor

###查看topic
/bin/kafka-topics.sh --bootstrap-server node05:9092 --describe --topic t_monitor
###  Topic: t_monitor  Partition: 0   Leader: 368   Replicas: 368,370,369  Isr: 368,370,369

###生产者
/kafka/bin/kafka-console-producer.sh --bootstrap-server node05:9092 --topic t_monitor

## 消费topic数据
/bin/kafka-console-consumer --bootstrap-server hostname:9092 --topic t_monitor --from-beginning 

##删除topic
 bin/kafka-topics.sh --bootstrap-server hostname:9092 --delete --topic first

##修改topic的分区数(只能增加不能减少)
bin/kafka-topics.sh --bootstrap-server hostname:9092 --alter --topic first --partitions 3





##查看消费者组的列表
bin/kafka-consumer-groups.sh --bootstrap-server node01:9092 --list

##查看消费组的消费详情
bin/kafka-consumer-groups.sh  --bootstrap-server node01:9092 --group test-01-group --describe 

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

1. kafka生产者

1. 生产者生产数据过程如下：

buffer.memory ：RecordAccumulator 缓冲区总大小，默认 32m
batch.size ：缓冲区一批数据最大值，默认 16k。适当增加该值，可以提高吞吐量，但是如果该值设置太大，会导致数据传输延迟增加
linger.ms ：如果数据迟迟未达到 batch.size，sender 等待 linger.time之后就会发送数据。单位 ms，默认值是 0ms，表示没有延迟。生产环境建议该值大小为 5-100ms 之间。
compression.type ：生产者发送的所有数据的压缩方式。默认是 none，也就是不压缩。支持压缩类型：none、gzip、snappy、lz4 和 zstd

2. 生产者发送数据分区策略
（1）指明partition的情况下，直接将指明的值作为partition值；例如partition=0，所有数据写入分区0
（2）没有指明partition值但有key的情况下，将key的hash值与topic的partition数进行取余得到partition值；
例如：key1的hash值=5， key2的hash值=6 ，topic的partition数=2，那 么key1 对应的value1写入1号分区，key2对应的value2写入0号分区。
（3）既没有partition值又没有key值的情况下，Kafka采用Sticky Partition（黏性分区器），会随机选择一个分区，并尽可能一直使用该分区，待该分区的batch已满或者已完成，Kafka再随机一个分区进行使用（和上一次的分区不同）。
例如：第一次随机选择0号分区，等0号分区当前批次满了（默认16k）或者linger.ms设置的时间到， Kafka再随机一个分区进
行使用（如果还是0会继续随机）。
（4）自定义分区器：实现Partitioner接口，实现 3 个方法:partition,close,configure 编写 partition 方法,返回分区号

3.生产者发送数据ACK应答级别
acks=0，生产者发送过来数据就不管了，可靠性差，效率高；
acks=1，生产者发送过来数据Leader应答，可靠性中等，效率中等；
acks=-1，生产者发送过来数据Leader和ISR队列里面所有Follwer应答，可靠性高，效率低；
在生产环境中，acks=0很少使用；acks=1，一般用于传输普通日志，允许丢个别数据；acks=-1，一般用于传输和钱相关的数据，
对可靠性要求比较高的场景。

properties.put(ProducerConfig.ACKS_CONFIG, "all"); --也是-1
1

2. Kafka Broker

1. Kafka Broker 总体工作流程

2. Broker 重要参数

##ISR 中，如果 Follower 长时间未向 Leader 发送通信请求或同步数据，则该 Follower 将被踢出 ISR。该时间阈值，默认 30s。
replica.lag.time.max.ms

##默认是 true。 自动 Leader Partition 平衡。
auto.leader.rebalance.enable 

##默认是 10%。每个 broker 允许的不平衡的 leader的比率。如果每个 broker 超过了这个值，控制器会触发 leader 的平衡。
leader.imbalance.per.broker.percentage 

##默认值 300 秒。检查 leader 负载是否平衡的间隔时间。
leader.imbalance.check.interval.seconds 

##Kafka 中 log 日志是分成一块块存储的，此配置是指 log 日志划分 成块的大小，默认值 1G。
log.segment.bytes 

##默认 4kb，kafka 里面每当写入了 4kb 大小的日志（.log），然后就往 index 文件里面记录一个索引。
log.index.interval.bytes 

##中数据保存的时间，默认 7 天。
log.retention.hours Kafka 

##中数据保存的时间，分钟级别，默认关闭。
log.retention.minutes Kafka 

##Kafka 中数据保存的时间，毫秒级别，默认关闭。
log.retention.ms 

## 检查数据是否保存超时的间隔，默认是 5 分钟。
log.retention.check.interval.ms

##默认等于-1，表示无穷大。超过设置的所有日志总大小，删除最早的 segment。
log.retention.bytes 

##默认是 delete，表示所有数据启用删除策略；如果设置值为 compact，表示所有数据启用压缩策略。
log.cleanup.policy 

##默认是 8。负责写磁盘的线程数。整个参数值要占总核数的 50%。 num.replica.fetchers 副本拉取线程数，这个参数占总核数的 50%的 1/3
num.io.threads 

##默认是 3。数据传输线程数，这个参数占总核数的50%的 2/3 。
num.network.threads 

## 强制页缓存刷写到磁盘的条数，默认是 long 的最大值，9223372036854775807。一般不建议修改，交给系统自己管理。
log.flush.interval.messages

##每隔多久，刷数据到磁盘，默认是 null。一般不建议修改，交给系统自己管理。
log.flush.interval.ms 
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

3. kafka消费者再平衡

## Kafka 消费者和 coordinator 之间的心跳时间，默认 3s。该条目的值必须小于 session.timeout.ms，也不应该高于session.timeout.ms 的 1/3。
heartbeat.interval.ms ： 3

## Kafka 消费者和 coordinator 之间连接超时时间，默认 45s。超过该值，该消费者被移除，消费者组执行再平衡。
session.timeout.ms :45

##消费者处理消息的最大时长，默认是 5 分钟。超过该值，该消费者被移除，消费者组执行再平衡。
max.poll.interval.ms : 5min

## 消费者分区策略 ： Range 、 RoundRobin 、 Sticky 、CooperativeSticky
partition.assignment.strategy ： Range + CooperativeSticky

1
2
3
4
5
6
7
8
9
10
11
12

若：消费者0挂

range模式：
消费者组需要按照超时时间 45s 来判断它是否退出，所以需要等待，时间到了 45s 后，判断它真的退出就会把任务分配给其他 broker 执行。消费者 0 已经被踢出消费者组，所以重新按照 range 方式分配。

RoundRobin ：模式
0 号消费者挂掉后，消费者组需要按照超时时间 45s 来判断它是否退出，所以需要等待，时间到了 45s 后，判断它真的退出就会把任务分配给其他 broker 执行。消费者 0 已经被踢出消费者组，所以重新按照 RoundRobin 方式分配