一、 Kafka常用操作命令
查看当前服务器中的所有topic
[root@hadoop3 kafka]# bin/kafka-topics.sh --list --zookeeper hadoop11:2181[root@hadoop3 kafka]#
信息写入到
创建topic
[root@hadoop3 kafka]# bin/kafka-topics.sh --create --zookeeper hadoop11:2181 --replication-factor 1 -partitions 1 --topic testCreated topic "test".[root@hadoop3 kafka]# bin/kafka-topics.sh --list --zookeeper hadoop11:2181test
通过上面,可以看到已经创建了一个test的topic
删除topic
[root@hadoop2 kafka]# bin/kafka-topics.sh --create --zookeeper hadoop11:2181 --replication-factor 1 -partitions 1 --topic test2Created topic "test2".[root@hadoop2 kafka]# bin/kafka-topics.sh --list --zookeeper hadoop11:2181itheimatesttest2[root@hadoop2 kafka]# bin/kafka-topics.sh --delete --zookeeper hadoop11:2181 --topic test2Topic test2 is marked for deletion.Note: This will have no impact if delete.topic.enable is not set to true.[root@hadoop2 kafka]# bin/kafka-topics.sh --list --zookeeper hadoop11:2181itheimatest[root@hadoop2 kafka]#
需要server.properties中设置delete.topic.enable=true否则只是标记删除或者直接重启。
通过shell命令发送消息
要注意的是要指定topic,表示要在哪个topic中生产消息,这里的topic需要时上面创建的topic
[root@hadoop3 kafka]# bin/kafka-console-producer.sh --broker-list hadoop1:9092 --topic testasdfasdfasdasdfasdfasdfasdftoto testtuto test2
注意命令中指定的–block-listhadoop1:9092,当改成hadoop2:9092时,也可以。
通过shell消费消息
要指明的是要使用哪个topic中的数据,这里的topic需要时上面创建的topic:
[root@hadoop3 kafka]# sh bin/kafka-console-consumer.sh --zookeeper hadoop11:2181 --from-beginning --topic testasdfasdfasdasdfasdfasdfasdftoto testtuto test2
注意:这里要指定消费那个topic,这里使用的是test.
查看消费位置
[root@hadoop3 kafka]# sh bin/kafka-run-class.sh kafka.tools.ConsumerOffsetChecker --zookeeper hadoop11:2181 --group testGroup
查看某个Topic的详情
[root@hadoop3 kafka]# sh bin/kafka-topics.sh --topic test --describe --zookeeper hadoop11:2181Topic:test PartitionCount:1 ReplicationFactor:1 Configs:Topic: test Partition: 0 Leader: 0 Replicas: 0 Isr: 0[root@hadoop3 kafka]#
对分区数进行修改
[root@hadoop3 kafka]# bin/kafka-topics.sh --zookeeper hadoop11:2181 -alter --partitions 15 --topic testWARNING: If partitions are increased for a topic that has a key, the partition logic or ordering of the messages will be affectedAdding partitions succeeded![root@hadoop3 kafka]#
查看当前服务器中的所有的topic 创建topic 删除topic 通过shell命令发送消息 通过shell消费消息 查看topic详情 对分区数进行修改