1、创建一个需要增加备份因子的topic列表的文件，文件格式是json格式的。

bin/kafka-create-topic.sh-replica 2-partition 8-topic test-zookeeper 192.168.197.170:2181,192.168.197.171:2181

创建名为test的topic，8个分区分别存放数据，数据备份总共2份。

2、使用kafka官方提供的工具拿到上面topic的partions 分布情况，并重定向到文件中。

3、修改ressgintopic.conf 文件的，手动分配新增加的partion 备份因子。

4、通过下面命令执行备份因子扩容过程，bin/kafka-reassign-partitions.sh --zookeeper localhost:2181 --reassignment-json。

5、最后查看kafka的某 topic数据如图。

bin/kafka-list-topic.sh-zookeeper 192.168.197.170:2181,192.168.197.171:2181 (列出所有topic的分区情况)

bin/kafka-list-topic.sh-zookeeper 192.168.197.170:2181,192.168.197.171:2181-topic test (查看test的分区情况)

1、创建一个需要增加备份因子的topic列表的文件，文件格式是json格式的。

2、使用kafka官方提供的工具拿到上面topic的partions 分布情况，并重定向到文件中。

3、修改ressgintopic.conf 文件的，手动分配新增加的partion 备份因子。

4、通过下面命令执行备份因子扩容过程，bin/kafka-reassign-partitions.sh --zookeeper localhost:2181 --reassignment-json。

5、最后查看kafka的某 topic数据如图。

创建一个需要增加备份因子的topic列表的文件，文件格式是json格式的。使用kafka官方提供的工具拿到上面topic的partions 分布情况，并重定向到文件中。

1.在命令行执行如下命令查看指定Topic及ConsumerGroup的消费情况[root@cdh01 bin]# kafka-run-class kafka.tools.ConsumerOffsetChecker --zookeeper cdh01.fayson.com:2181 --group test --topic kafka_kudu_topic_offset_3第一次执行查看到kafka_kudu_topic_offset_3的testgroup组各个Partition的Offset消费情况以及每个Partition的总消息量。

1、查看topic分布情况kafka-list-topic.sh：

bin/kafka-list-topic.sh-zookeeper 192.168.197.170:2181,192.168.197.171:2181 （列出所有topic的分区情况）

bin/kafka-list-topic.sh-zookeeper 192.168.197.170:2181,192.168.197.171:2181-topic test （查看test的分区情况）

2、创建topickafka-create-topic.sh：bin/kafka-create-topic.sh-replica 2-partition 8-topic test-zookeeper 192.168.197.170:2181,192.168.197.171:2181

3、创建名为test的topic，8个分区分别存放数据，数据备份总共2份。

1.文档编写目的

在我们的应用程序消费Kafka的Topic消息时，我们会关注Topic中消息总量以及消费情况，那接下来Fayson就主要介绍下如何在CDH集群中使用命令行的方式查看Topic的消费情况。

本次Fayson的测试环境为

1.CM和CDH版本为5.15

2.Redhat7.4

3.Kafka 0.10.2+kafka2.2.0

2.命令行查看

在Kafka的脚本里提供了kafka-run-class.sh的脚本，使用该脚本查看Consumer组内消费Topic的Offset情况。

1.在命令行执行如下命令查看指定Topic及ConsumerGroup的消费情况

[root@cdh01 bin]# kafka-run-class kafka.tools.ConsumerOffsetChecker --zookeeper cdh01.fayson.com:2181 --group test --topic kafka_kudu_topic_offset_3

第一次执行查看到kafka_kudu_topic_offset_3的testgroup组各个Partition的Offset消费情况以及每个Partition的总消息量。

2.kafka_kudu_topic_offset_3的Topic发送600条消息，再次执行命令查看消费情况

此时没有消费的应用运行，可以看到Offset的数据未发生变化，而logSize的数据量有增加，每个partition的数量增加200条，刚好为我发送的消息数量。

3.启动消费程序消费Topic的消息，再次执行命令查看消费情况

在消费应用运行时，每个Partition的Offset数据发生变化，此时的Offset的数据与logSize的数据一致，在应用日志中也无新的消息接收，说明Topic的消息已被testgroup组消费完。

3.总结

查看某个Topic指定组的消息消费情况计算方式：

消费的消息量计算方式：

topic_partition1 offset + topic_partition2 offset +… + topic_partition_n offset

Topic总消息数量计算方式：

topic_partition1 logSize + topic_partition2 logSize +… + topic_partition_n logSize

未消费消息数量计算方式：

Topic总消息数量-消费的消息量

提示：代码块部分可以左右滑动查看噢

• 查看topic分布情况kafka-list-topic.sh：

  bin/kafka-list-topic.sh-zookeeper 192.168.197.170:2181,192.168.197.171:2181 （列出所有topic的分区情况）

  bin/kafka-list-topic.sh-zookeeper 192.168.197.170:2181,192.168.197.171:2181-topic test （查看test的分区情况）

• 创建topickafka-create-topic.sh：

  bin/kafka-create-topic.sh-replica 2-partition 8-topic test-zookeeper 192.168.197.170:2181,192.168.197.171:2181

  创建名为test的topic，8个分区分别存放数据，数据备份总共2份。

1、查询topic，进入kafka目录：

bin/kafka-topics.sh --list --zookeeper localhost:2181

2、查询topic内容：

bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic topicName --from-beginning