Kafka 脚本补全

2022年5月6日 · 4281 字 · 9 分钟 · Kafka

1、概览

kafka 安装目录下的 bin 目录包含了很多运维可操作的 shell 脚本，列举如下：

脚本名称	用途描述
connect-distributed.sh	连接 kafka 集群模式
connect-standalone.sh	连接 kafka 单机模式
kafka-acls.sh	todo
kafka-broker-api-versions.sh	todo
kafka-configs.sh	配置管理脚本
kafka-console-consumer.sh	kafka 消费者控制台
kafka-console-producer.sh	kafka 生产者控制台
kafka-consumer-groups.sh	kafka 消费者组相关信息
kafka-consumer-perf-test.sh	kafka 消费者性能测试脚本
kafka-delegation-tokens.sh	todo
kafka-delete-records.sh	删除低水位的日志文件
kafka-log-dirs.sh	kafka 消息日志目录信息
kafka-mirror-maker.sh	不同数据中心 kafka 集群复制工具
kafka-preferred-replica-election.sh	触发 preferred replica 选举
kafka-producer-perf-test.sh	kafka 生产者性能测试脚本
kafka-reassign-partitions.sh	分区重分配脚本
kafka-replay-log-producer.sh	todo
kafka-replica-verification.sh	复制进度验证脚本
kafka-run-class.sh	todo
kafka-server-start.sh	启动 kafka 服务
kafka-server-stop.sh	停止 kafka 服务
kafka-simple-consumer-shell.sh	deprecated，推荐使用 kafka-console-consumer.sh
kafka-streams-application-reset.sh	todo
kafka-topics.sh	topic 管理脚本
kafka-verifiable-consumer.sh	可检验的 kafka 消费者
kafka-verifiable-producer.sh	可检验的 kafka 生产者
trogdor.sh	todo
zookeeper-security-migration.sh	todo
zookeeper-server-start.sh	启动 zk 服务
zookeeper-server-stop.sh	停止 zk 服务
zookeeper-shell.sh	zk 客户端

2、connect-distributed.sh&connect-standalone.sh

Kafka Connect 是在 0.9 以后加入的功能，主要是用来将其他系统的数据导入到 Kafka, 然后再将 Kafka 中的数据导出到另外的系统。可以用来做实时数据同步的 ETL，数据实时分析处理等。主要有 2 种模式：Standalone（单机模式）和 Distribute（分布式模式）。单机主要用来开发，测试，分布式的用于生产环境。

3、kafka-broker-api-versions.sh

bin/kafka-broker-api-versions.sh --bootstrap-server 10.0.55.229:9092,10.0.55.229:9093,10.0.55.229:9094

4、kafka-configs.sh

配置管理脚本，这个脚本主要分两类用法：describe 和 alter。 describe 相关用法：查看每个 topic 的配置：

sh kafka-configs.sh --bootstrap-server 127.0.0.1:9092 --describe --entity-type topics

部分结果如下：

Configs for topic 'afei' are 
Configs for topic 'TOPIC-TEST-AFEI' are retention.ms=600000 
Configs for topic '__consumer_offsets' are segment.bytes=104857600,cleanup.policy=compact,compression.type=producer

查看 broker 的配置：

bin/kafka-configs.sh --bootstrap-server localhost:9092 --describe --entity-type brokers --entity-name 0

说明：0 是 broker.id，因为 entity-type 为 brokers，所以 entity-name 表示 broker.id。

alter 相关用法：给指定 topic 增加配置

sh kafka-configs.sh --zookeeper localhost:2181 --alter --entity-type topics --entity-name TOPIC-TEST-AFEI --add-config retention.ms=600000

给指定 topic 删除配置：

sh kafka-configs.sh --zookeeper localhost:2181 --alter --entity-type topics --entity-name TOPIC-TEST-AFEI --delete-config max.message.bytes

通过该脚本可以管理的属性，可以通过执行 bin/kafka-configs.sh 得到的输出中 –add-config 的 desc 可以得到。

5、kafka-broker-api-versions.sh

bin/kafka-broker-api-versions.sh --bootstrap-server localhost:9092

6、kafka-console-consumer.sh

用法：

bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic afei [--group groupName] [--partition 目标分区]

这个命令后面还可带很多参数：

--key-deserializer：指定 key 的反序列化方式，默认是 org.apache.kafka.common.serialization.StringDeserializer
--value-deserializer：指定 value 的反序列化方式，默认是 org.apache.kafka.common.serialization.StringDeserializer
--from-beginning：从最早的消息开始消费，默认是从最新消息开始消费。
--offset：从指定的消息位置开始消费，如果设置了这个参数，还需要带上 –partition。否则会提示：The partition is required when offset is specified.
--timeout-ms：当消费者在这个参数指定时间间隔内没有收到消息就会推出，并抛出异常：kafka.consumer.ConsumerTimeoutException。
--whitelist：接收的 topic 白名单集合，和 –topic 二者取其一。例如：–whitelist “afei.*"（以 afei 开头的 topic），–whitelist “afei”（指定 afei 这个 topic），“afei|fly”（指定 afei 或者 fly 这两个 topic）。另外一个参数 –blacklist 用法类似。

7、kafka-console-producer.sh

用法：

bin/kafka-console-producer.sh --broker-list localhost:9092 --topic <topic_name>

，如果连接集群，那么 broker-list 参数格式为：HOST1:PORT1,HOST2:PORT2,HOST3:PORT3 这个命令后面还可带很多参数：

--key-serializer：指定 key 的序列化方式，默认是 org.apache.kafka.common.serialization.StringSerializer
--value-serializer：指定 value 的序列化方式，默认是 org.apache.kafka.common.serialization.StringSerializer

8、kafka-consumer-groups.sh

查看所有消费者组：bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --list 查看某个消费者组：bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --group <group_name>--describe，输出结果如下：

9、kafka-consumer-perf-test.sh

perf 是 performance 的缩写，所以这个脚本是 kafka 消费者性能测试脚本。用法：

bin/kafka-consumer-perf-test.sh --broker-list localhost:9092 --group testGroup --topic afei --messages 1024

输出结果如下：

start.time, end.time, data.consumed.in.MB, MB.sec, data.consumed.in.nMsg, nMsg.sec, rebalance.time.ms, fetch.time.ms, fetch.MB.sec, fetch.nMsg.sec
2018-07-02 22:49:10:068, 2018-07-02 22:49:12:077, 0.0001, 0.0001, 41, 20.4082, 19, 1990, 0.0001, 20.6030

10、kafka-delete-records.sh

用法：

bin/kafka-delete-records.sh --bootstrap-server 10.0.55.229:9092,10.0.55.229:9093,10.0.55.229:9094 --offset-json-file offset.json

offset.json 文件内容：

{
    "partitions": [{
        "topic": "<topic_name>",
        "partition": 3,
        "offset": 10
    }],
    "version": 1
}

执行结果如下，表示删除 afei 这个 topic 下分区为 3 的 offset 少于 10 的消息日志（不会删除 offset=10 的消息日志）：

Executing records delete operation
Records delete operation completed:
partition: afei-3   low_watermark: 10

11、kafka-log-dirs.sh

用法：

bin/kafka-log-dirs.sh --bootstrap-server localhost:9092 --describe --topic-list [topicName1,topicName2,topicNameN]

如果没有指定 –topic-list，那么会输出所有 kafka 消息日志目录以及目录下所有 topic 信息。加上 –topic-list 参数后，输出结果如下，由这段结果可知，消息日志所在目录为 /data/kafka-logs，并且 afei 这个 topic 有 3 个分区：

12、kafka-preferred-replica-election.sh

用法：

bin/kafka-preferred-replica-election.sh --zookeeper 10.0.55.208:2181/wallet,10.0.55.209:2181/wallet,10.0.55.210:2181/wallet --path-to-json-file afei-preferred.json

（如果不带 –path-to-json-file 就是对所有 topic 进行 preferred replica election），json 文件内容如下：：

{
    "partitions": [{
        "topic": "afei",
        "partition": 0
    },
    {
        "topic": "afei",
        "partition": 1
    },
    {
        "topic": "afei",
        "partition": 2
    }]
}

场景：在创建一个 topic 时，kafka 尽量将 partition 均分在所有的 brokers 上，并且将 replicas 也均分在不同的 broker 上。每个 partitiion 的所有 replicas 叫做 “assigned replicas”，“assigned replicas” 中的第一个 replicas 叫 “preferred replica”，刚创建的 topic 一般 “preferred replica” 是 leader。leader replica 负责所有的读写。其他 replica 只是冷备状态，不接受读写请求。但随着时间推移，broker 可能会主动停机甚至客观宕机，会发生 leader 选举迁移，导致机群的负载不均衡。我们期望对 topic 的 leader 进行重新负载均衡，让 partition 选择 “preferred replica” 做为 leader。 kafka 提供了一个参数 auto.leader.rebalance.enable 自动做这件事情，且默认为 true，原理是一个后台线程检查并触发 leader balance。但是并不建议把这个参数设置为 true。因为担心这个自动选举发生在业务高峰期，从而导致影响业务。验证：操作比较简单，常见一个 3 个分区 3 个副本的 topic，然后 kill 掉一个 broker。这时候 topic 信息如下，我们可以看到 broker.id 为 0 的 broker 上有两个 leader：

Topic:afei  PartitionCount:3    ReplicationFactor:3 Configs:
    Topic: afei Partition: 0    Leader: 0   Replicas: 0,1,2 Isr: 0,1,2
    Topic: afei Partition: 1    Leader: 1   Replicas: 1,2,0 Isr: 0,1,2
    Topic: afei Partition: 2    Leader: 0   Replicas: 2,0,1 Isr: 0,1,2

执行 kafka-preferred-replica-election.sh 脚本后，topic 信息如下，leader 均匀分布在 3 个不同的 broker 上，

Topic:afei  PartitionCount:3    ReplicationFactor:3 Configs:
    Topic: afei Partition: 0    Leader: 0   Replicas: 0,1,2 Isr: 0,1,2
    Topic: afei Partition: 1    Leader: 1   Replicas: 1,2,0 Isr: 0,1,2
    Topic: afei Partition: 2    Leader: 2   Replicas: 2,0,1 Isr: 0,1,2

13、kafka-producer-perf-test.sh

perf 是 performance 的缩写，所以这个脚本是 kafka 生产者性能测试脚本。

14、kafka-reassign-partitions.sh

场景：将一些 topic 上的分区从当前所在 broker 移到其他比如新增的 broker 上。假设有个名为 ORDER-DETAIL 的 topic，在 broker.id 为 2 的 broker 上：

Topic:ORDER-DETAIL  PartitionCount:1    ReplicationFactor:1 Configs:
    Topic: ORDER-DETAIL Partition: 0    Leader: 2   Replicas: 2 Isr: 2

现在想要把它移动到 broker.id 为 1 的 broker 上，执行脚本：

bin/kafka-reassign-partitions.sh --zookeeper 10.0.55.208:2181/wallet,10.0.55.209:2181/wallet,10.0.55.210:2181/wallet --topics-to-move-json-file move.json --broker-list "1" --generate

--generate 参数表示生成一个分区再分配配置，并不会真正的执行，命令执行结果如下：

Current partition replica assignment
{"version":1,"partitions":[{"topic":"ORDER-DETAIL","partition":0,"replicas":[2],"log_dirs":["any"]}]}
 
Proposed partition reassignment configuration
{"version":1,"partitions":[{"topic":"ORDER-DETAIL","partition":0,"replicas":[1],"log_dirs":["any"]}]}

我们只需要把第二段 json 内容保存到一个新建的 final.json 文件中（如果知道如何编写这段 json 内容，那么也可以不执行第一条命令），然后执行命令：

bin/kafka-reassign-partitions.sh --zookeeper 10.0.55.208:2181/wallet,10.0.55.209:2181/wallet,10.0.55.210:2181/wallet --reassignment-json-file move_final.json --execute

此次执行的命令带有 –execute 参数，说明是真正的执行分区重分配。通过这个命令还可以给某个 topic 增加副本，例如有一个名为 ORDER-DETAIL 的 topic，有 3 个分区，但是只有 1 个副本，为了高可用，需要将副本数增加到 2，那么编写 replica.json 文本内容如下：

{
    "version": 1,
    "partitions": [{
        "topic": "ORDER-DETAIL",
        "partition": 0,
        "replicas": [1, 2]
    },
    {
        "topic": "ORDER-DETAIL",
        "partition": 1,
        "replicas": [0, 2]
    },
    {
        "topic": "ORDER-DETAIL",
        "partition": 2,
        "replicas": [1, 0]
    }]
}

然后执行命令即可：

bin/kafka-reassign-partitions.sh --zookeeper 10.0.55.208:2181/wallet,10.0.55.209:2181/wallet,10.0.55.210:2181/wallet --reassignment-json-file replica.json

15、kafka-replica-verification.sh

用法：

bin/kafka-replica-verification.sh --broker-list 10.0.55.229:9092,10.0.55.229:9093,10.0.55.229:9094 [--topic-white-list afei]

参数 --topic-white-list 指定要检查的目标 topic。输出结果如下，如果输出信息为 max lag is 0 for … 表示这个 topic 的复制没有任何延迟：

2023-07-03 15:04:46,889: verification process is started.
2023-07-03 15:05:16,811: max lag is 0 for partition multi-1 at offset 0 among 5 partitions
2023-07-03 15:05:46,812: max lag is 0 for partition multi-1 at offset 0 among 5 partitions
... ...

16、kafka-server-start.sh

bin/kafka-server-start.sh -daemon config/server.properties

指定配置文件并以守护进程模式启动。

17、kafka-server-stop.sh

用法：bin/kafka-server-stop.sh。说明，这个命令会 kill 掉当前服务器上所有 kafka broker。但是这个脚本可能执行结果为：No kafka server to stop。分析原因：先看一下 kafka-server-stop.sh 脚本内容，这个脚本非常简单，就是得到所有包含 kafka.Kafka 的进程 ID，但是由于 kafka 启动依赖比较多的 jar，导致 kafka 进程的 ps 结果输出内容比较长，而 ps 输出结果受到 PAGE_SIZE（其值通过命令 getconf PAGE_SIZE 可以得到）的限制，从而导致 ps 结果中看不到 kafka.Kafka，所以不能 kill 掉 kafka server：

SIGNAL=${SIGNAL:-TERM}
PIDS=$(ps ax | grep -i 'kafka\.Kafka' | grep java | grep -v grep | awk '{print $1}')
 
if [ -z "$PIDS" ]; then
  echo "No kafka server to stop"
  exit 1
else
  kill -s $SIGNAL $PIDS
fi

为了 kafka-server-stop.sh 脚本可以正常执行，建议修改脚本如下，通过 bin 脚本所在目录的上级目录来查找进程 ID，从而 kill 相关进程：

cd `dirname $0`
BIN_DIR=`pwd`
cd ..
DEPLOY_DIR=`pwd`
SIGNAL=${SIGNAL:-TERM}
 
PIDS=$(ps ax | grep -i "${DEPLOY_DIR}" | grep java | grep -v grep | awk '{print $1}')
 
if [ -z "$PIDS" ]; then
  echo "No kafka server to stop"
  exit 1
else
  kill -s $SIGNAL $PIDS
fi

18、kafka-simple-consumer-shell.sh

用法：

bin/kafka-simple-consumer-shell.sh --broker-list 10.0.55.229:9092,10.0.55.229:9093,10.0.55.229:9094 --topic afei

19、kafka-topics.sh

创建 topic:

bin/kafka-topics.sh --zookeeper localhost:2181 --create --topic <topic_name> --partitions 3 --replication-factor 1

删除 topic:

bin/kafka-topics.sh --zookeeper localhost:2181 --delete --topic <topic_name>

修改 topic:

bin/kafka-topics.sh --zookeeper localhost:2181 --alter --topic <topic_name> --partitions 5

修改 topic 时只能增加分区数量。查询 topic:

bin/kafka-topics.sh --zookeeper localhost:2181 --describe <topic_name>

查询时如果带上 –topic topicName，那么表示只查询该 topic 的详细信息。这时候还可以带上 –unavailable-partitions 和 –under-replicated-partitions 任意一个参数。创建 TOPIC：

./kafka-topics.sh  --create --zookeeper centos_1:2181  --replication-factor 1 --partitions 1 --topic <topic_name>

查看 TOPIC 是否创建成功：

./kafka-topics.sh --list --zookeeper centos_1:2181 <topic_name>

20、kafka-verifiable-consumer.sh

bin/kafka-verifiable-consumer.sh --broker-list 10.0.55.229:9092,10.0.55.229:9093,10.0.55.229:9094 --topic <topic_name> --group-id <group_name>

这个脚本的作用是接收指定 topic 的消息消费，并发出消费者事件，例如：offset 提交等。

21、kafka-verifiable-producer.sh

bin/kafka-verifiable-producer.sh --broker-list 10.0.55.229:9092,10.0.55.229:9093,10.0.55.229:9094 --topic <topic_name> [--max-messages 64]

建议使用该脚本时增加参数 –max-messages，否则会不停的发送消息。这个脚本的作用是持续发送消息到指定的 topic 中，参数 –max-messages 限制最大发送消息数。且每条发送的消息都会有响应信息，这就是和 kafka-console-producer.sh 最大的不同：

22、zookeeper-shell.sh

bin/zookeeper-shell.sh localhost:2181 [/path]

如果 kafka 集群的 zk 配置了 chroot 路径，那么需要加上 /path，例如 bin/zookeeper-shell.sh localhost:2181/mykafka，登陆 zk 后，就可以查看 kafka 写在 zk 上的节点信息。例如查看有哪些 broker，以及 broker 的详细信息：

ls /brokers/ids
[0]
get /brokers/ids/0
{"listener_security_protocol_map":{"PLAINTEXT":"PLAINTEXT"},"endpoints":["PLAINTEXT://izwz91rhzhed2c54e6yx87z:9092"],"jmx_port":-1,"host":"izwz91rhzhed2c54e6yx87z","timestamp":"1530435834272","port":9092,"version":4}
cZxid = 0x2d3
ctime = Sun Jul 01 17:03:54 CST 2018
mZxid = 0x2d3
mtime = Sun Jul 01 17:03:54 CST 2018
pZxid = 0x2d3
cversion = 0
dataVersion = 0
aclVersion = 0
ephemeralOwner = 0x1642cb09421006c
dataLength = 216
numChildren = 0

Kafka 数据迁移 Kafka 安全(SASL+ACL)