悄悄掌握 Kafka 常用命令,再也不用全网搜索了(建议收藏)

?

悄悄掌握 Kafka 常用命令,再也不用全网搜索了(建议收藏)

每一个成功人士的背后,必定曾经做出过勇敢而又孤独的决定。

放弃不难,但坚持很酷~

前言

对于从事大数据相关职位的朋友们来说,使用 kafka 的频率应该不会少。为了解决各位在操作 kafka 时记不住命令参数的痛点,所以我整理了一下在我工作中经常用到的 kafka 实操命令,希望各位看官能够喜欢。

?

kafka版本:2.11-1.1.0

一、kafka shell 命令行汇总

1、查看当前的集群Topic列表

./bin/kafka-topics.sh?--list?--zookeeper?cdh-worker-1:2181/kafka

2、查看所有的Topic的详细信息

./bin/kafka-topics.sh?--describe?--zookeeper?cdh-worker-1:2181/kafka

如果要查看单个 topic 信息:可在上述命令后面添加 --topic <topicName>

3、创建Topic

./bin/kafka-topics.sh?--create?--zookeeper?cdh-worker-1:2181/kafka?--replication-factor?3?--partitions?1?--topic?test-topic

4、删除Topic

删除 topic 之前,需要确保配置 delete.topic.enable=true 。

./bin/kafka-topics.sh?--delete?--zookeeper?cdh-worker-1:2181/kafka?--topic?topic-demo
Topic?topic-demo?is?marked?for?deletion.
Note:?This?will?have?no?impact?if?delete.topic.enable?is?not?set?to?true.

执行完命令后,查看 log.dirs 指定的文件目录,会发现 topic-demo 的文件夹都被标记为 delete ,如下图所示。

悄悄掌握 Kafka 常用命令,再也不用全网搜索了(建议收藏)

等一定的时间(根据 log.retention.check.interval.ms 配置而定,hdp 版本默认为 60s)后,被标记为 delete 的文件则会被移除。

5、生产数据

./bin/kafka-console-producer.sh?--broker-list?kafka-1:9092?--topic?test-topic
>?This?is?a?messageThis?is?another?message

6、消费数据

./bin/kafka-console-consumer.sh?--bootstrap-server?kafka-1:9092?--topic?test-topic?--from-beginning

--from-beginning 表示从最初的未过期的 offset 处开始消费数据。不加该参数,表示从最新 offset 处开始消费数据。

7、查询topic的offect范围

查询offect的最小值:

./bin/kafka-run-class.sh?kafka.tools.GetOffsetShell?--broker-list?kafka-1:9092?-topic?test-topic?--time?-2
#?输出
test-topic:0:0

查询offect的最大值:

./bin/kafka-run-class.sh?kafka.tools.GetOffsetShell?--broker-list?192.168.78.184:9092?-topic?test-topic?[--time?-1]
#?输出
test-topic:0:655

从上面的输出可以看出 test-topic 只有一个 Partition:0;offset 的范围是【0,655】。

8、增加分区

将分区数增加到 3 个:

./bin/kafka-topics.sh?--alter?--zookeeper?cdh-worker-1:2181/kafka?--topic?test-topic?--partitions?3

如果需要重新分布 kafka 分区以及增加分区副本数,可以参考:《必会 | 教你如何重新分布kafka分区、增加分区副本数》

9、均衡 kafka 的 leader 副本

可以参考我之前写的干货文章:《kafka 如何对 topic 分区 replica leader 进行负载均衡》

10、查看消费组

./bin/kafka-consumer-groups.sh?--bootstrap-server?kafka-1:9092?--list

查看指定消费组的详情(比如消费进度 LAG ),这里的消费者组名为 console-consumer-3665 :

./bin/kafka-consumer-groups.sh?--bootstrap-server?kafka-1:9092?--group?console-consumer-3665??--describe

11、指定 partition 和 offset 消费

./bin/kafka-console-consumer.sh?--bootstrap-server?kafka-1:9092?--topic?test-topic?--partition?0?--offset?1663520

12、从__consumer_offsets主题查找某个group的偏移量

1)计算 group.id 对应的 partition

__consumer_offsets 默认有 50 个 partition ,需要先计算 group.id 对应的 partition ,计算公式如下所示:

#?计算公式
Math.abs(groupid.hashCode())?%?numPartitions
#?实例,groupid 为 console-consumer-3665,numPartitions 是 50。
Math.abs("console-consumer-3665".hashCode())?%?50
#?得到的数字,就是你消费者组对应的 partition 。

2)消费分区

找到 partition 后,就可以消费指定分区了:

./bin/kafka-console-consumer.sh?\
--bootstrap-server?kafka-1:9092?\
--topic?__consumer_offsets?\
--formatter?"kafka.coordinator.group.GroupMetadataManager\$OffsetsMessageFormatter"?\
--partition?17?|?grep?xxx

注意事项

在 kafka 0.11.0.0 版本之前 --formatter 需要使用 kafka.coordinator.GroupMetadataManager\$OffsetsMessageFormatter,0.11.0.0 版本以后(含)使用上面脚本中使用的 Class 。

13、为 topic 设置单独配置

为 test-topic 设置某配置参数。

./bin/kafka-configs.sh?--zookeeper?cdh-worker-1:2181/kafka??--entity-type?topics?--entity-name?test-topic?--alter?--add-config?max.message.bytes=10485760?

查看这个 topic 设置的参数:

./bin/kafka-configs.sh?--zookeeper?cdh-worker-1:2181/kafka??--entity-type?topics?--entity-name?test-topic?--describe?

14、查看 kafla 数据 xxx.log 日志

./bin/kafka-run-class.sh?kafka.tools.DumpLogSegments?--files?/data/kafka_data/logs/test-0/00000000000001049942.log?--print-data-log?--deep-iteration?>?secLog.log

二、小结

上面是我一直以来积累的 kafka 常用命令,挺齐全的了。指定 partition 和 offset 消费数据、查看消费者组消费情况,查看消费者组的提交 offset 信息,增加分区、均衡分区、增加分区副本数、均衡 leader 副本等等。

为了解决各位在操作 kafka 时记不住命令参数的痛点,所以贴心的我整理了一下在我工作中经常用到的 kafka 实操命令,希望各位看官能够喜欢。

?

往期推荐

https://blog.51cto.com/u_13193683/2977129

?

悄悄掌握 Kafka 常用命令,再也不用全网搜索了(建议收藏)

?

悄悄掌握 Kafka 常用命令,再也不用全网搜索了(建议收藏)

上一篇:.net 直接输出远程文件到浏览器和下载文件保存到本机


下一篇:ERROR: Cannot create ipc data