Kafka Shell基本命令（包括topic的增删改查）

雨点打透心脏的1/2处 2022-05-17 11:05 172阅读 0赞

**创建kafka topic**

bin/kafka-topics.sh --zookeeper node01:2181 --create --topic t\_cdr --partitions 30 --replication-factor 2

注： partitions指定topic分区数，replication-factor指定topic每个分区的副本数

*  partitions分区数:

*   *  partitions ：分区数，控制topic将分片成多少个log。可以显示指定，如果不指定则会使用broker(server.properties)中的num.partitions配置的数量
     *  虽然增加分区数可以提供kafka集群的吞吐量、但是过多的分区数或者或是单台服务器上的分区数过多，会增加不可用及延迟的风险。因为多的分区数，意味着需要打开更多的文件句柄、增加点到点的延时、增加客户端的内存消耗。
     *  分区数也限制了consumer的并行度，即限制了并行consumer消息的线程数不能大于分区数
     *  分区数也限制了producer发送消息是指定的分区。如创建topic时分区设置为1，producer发送消息时通过自定义的分区方法指定分区为2或以上的数都会出错的；这种情况可以通过alter –partitions 来增加分区数。

*  replication-factor副本

*   *  replication factor 控制消息保存在几个broker(服务器)上，一般情况下等于broker的个数。
     *  如果没有在创建时显示指定或通过API向一个不存在的topic生产消息时会使用broker(server.properties)中的default.replication.factor配置的数量

**查看所有topic列表**

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

**查看指定topic信息**

bin/kafka-topics.sh --zookeeper node01:2181 --describe --topic t\_cdr

**控制台向topic生产数据**

bin/kafka-console-producer.sh --broker-list node86:9092 --topic t\_cdr

**控制台消费topic的数据**

bin/kafka-console-consumer.sh --zookeeper node01:2181 --topic t\_cdr --from-beginning

**查看topic某分区偏移量最大（小）值**

bin/kafka-run-class.sh kafka.tools.GetOffsetShell --topic hive-mdatabase-hostsltable --time -1 --broker-list node86:9092 --partitions 0

注： time为-1时表示最大值，time为-2时表示最小值

**增加topic分区数**

为topic t\_cdr 增加10个分区

bin/kafka-topics.sh --zookeeper node01:2181 --alter --topic t\_cdr --partitions 10

**删除topic，慎用，只会删除zookeeper中的元数据，消息文件须手动删除**

bin/kafka-run-class.sh kafka.admin.DeleteTopicCommand --zookeeper node01:2181 --topic t\_cdr

1.  ./kafka-topics.sh --delete --zookeeper Slave1:2181 --topic kafkatopic5  
2.  Topic kafkatopic5 is marked for deletion.  
3.  Note: This will have no impact if delete.topic.enable is not set to true.

用上面这种方法删除主题后，有一个提示信息，如下所示

\[html\] [view plain][] [copy][view plain]

1.  <pre name="code" class="html">Topic kafkatopic5 is marked for deletion.

\[html\] [view plain][] [copy][view plain]

1.  <span style="font-family: Arial, Helvetica, sans-serif; font-size: 12px;">Note: This will have no impact if delete.topic.enable is not set to true.</span>

要删除的kafkatopic5被标记为删除，如果delete.topic.enble没有被标记为true，它将不会产生影响。

需要配置delete.topic.enble

\[html\] [view plain][] [copy][view plain]

1.  \[hadoop@Slave1 bin\]$ cd ..  
2.  \[hadoop@Slave1 kafka\]$ ls  
3.  bin  config  libs  LICENSE  logs  NOTICE  
4.  \[hadoop@Slave1 kafka\]$ cd config/  
5.  \[hadoop@Slave1 config\]$ ls  
6.  consumer.properties  server.properties       zookeeper.properties  
7.  log4j.properties     test-log4j.properties  
8.  producer.properties  tools-log4j.properties  
9.  \[hadoop@Slave1 config\]$ vim server.properties   
10. \[hadoop@Slave1 config\]$

对config目录下的server.properties进行修改，在最末行加上：delete.topic.enble=true

在Slave2和Slave3上做同样修改

\[html\] [view plain][] [copy][view plain]

1.  \[hadoop@Slave1 config\]$ ssh Slave2  
2.  Last login: Wed Nov 11 17:41:53 2015 from master  
3.  \[hadoop@Slave2 ~\]$ cd /usr/local/kafka/config/  
4.  \[hadoop@Slave2 config\]$ vim server.properties   
5.  \[hadoop@Slave2 config\]$ exit  
6.  logout  
7.  Connection to Slave2 closed.  
8.  \[hadoop@Slave1 config\]$ ssh Slave3  
9.  Last login: Wed Nov 11 17:42:32 2015 from master  
10. \[hadoop@Slave3 ~\]$ cd /usr/local/kafka/config/  
11. \[hadoop@Slave3 config\]$ vim server.properties   
12. \[hadoop@Slave3 config\]$ exit  
13. logout  
14. Connection to Slave3 closed.  
15. \[hadoop@Slave1 config\]$

**查看topic消费进度**

这个会显示出consumer group的offset情况， 必须参数为--group， 不指定--topic，默认为所有topic

Displays the: Consumer Group, Topic, Partitions, Offset, logSize, Lag, Owner for the specified set of Topics and Consumer Group

bin/kafka-run-class.sh kafka.tools.ConsumerOffsetChecker required argument: \[group\] Option Description ------ ----------- --broker-info Print broker info --group Consumer group. --help Print this message. --topic Comma-separated list of consumer topics (all topics if absent). --zkconnect ZooKeeper connect string. (default: localhost:2181) Example, bin/kafka-run-class.sh kafka.tools.ConsumerOffsetChecker --group pv Group Topic Pid Offset logSize Lag Owner pv page\_visits 0 21 21 0 none pv page\_visits 1 19 19 0 none pv page\_visits 2 20 20 0 none

[view plain]: http://blog.csdn.net/wang_zhenwei/article/details/49795981#