1、集群管理
前台啟動broker
bin/kafka-server-start.sh <path>/server.properties
Ctrl + C 關閉
后台啟動broker
bin/kafka-server-start.sh -daemon <path>/server.properties
關閉broker
bin/kafka-server-stop.sh
2、Topic管理
創建topic
bin/kafka-topics.sh --create --zookeeper localhost:2181 --partitions 3 --replication-factor 3 --topic topicname
刪除topic
bin/kafka-topics.sh --delete --zookeeper localhost:2181 --topic topicname
查詢topic列表
bin/kafka-topics.sh --zookeeper localhost:2181 --list
查詢topic詳情
bin/kafka-topics.sh --zookeeper localhost:2181 --describe --topic topicname
修改topic
bin/kafka-topics.sh --alter --zookeeper localhost:2181 --partitions 6 --topic topicname
3、Consumer-Groups管理
查詢消費者組
bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --list
查詢消費者組詳情
bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --describe --group groupname
重設消費者組位移
最早處
bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --group groupname --reset-offsets --all-topics --to-earliest --execute
最新處
bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --group groupname --reset-offsets --all-topics --to-latest --execute
某個位置
bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --group groupname --reset-offsets --all-topics --to-offset 2000 --execute
調整到某個時間之后得最早位移
bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --group groupname --reset-offsets --all-topics --to-datetime 2019-09-15T00:00:00.000
刪除消費者組
bin/kafka-consumer-groups.sh --zookeeper localhost:2181 --delete --group groupname
4、腳本工具
producer腳本
bin/kafka-console-producer.sh --broker-list localhost:9092 --topic topicname
參數含義：
--compression-codec lz4 壓縮類型
--request-required-acks all acks的值
--timeout 3000 linger.ms的值
--message-send-max-retries 10 retries的值
--max-partition-memory-bytes batch.size值
consumer腳本
bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic topicname --from-beginning
指定groupid
bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic topicname --from-beginning
--consumer-property group.id=old-consumer-group
指定分區
bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic topicname --from-beginning
--partition 0
kafka-run-class腳本
kafka-run-class.sh kafka.tools.ConsoleConsumer 就是 kafka-console-consumer.sh
kafka-run-class.sh kafka.tools.ConsoleProducer 就是 kafka-console-producer.sh
獲取topic當前消息數
kafka-run-class.sh kafka.tools.GetOffsetShell --broker-list localhost:9092 --topic topicname --time -1
--time -1表示最大位移 --time -2表示最早位移
查詢_consumer_offsets
bin/kafka-simple-consumer-shell.sh --topic _consumer_offsets --partition 12 --broker-list localhost:9092 --formatter "kafka.coorfinator.GroupMetadataManager\$OffsetsMessageFormatter"
5、MirrorMaker
跨機房災備工具
bin/kafka-mirror-maker.sh --consumer.config consumer.properties --producer.config producer.properties --whitelist topicA|topicB
更多實時計算,Flink,Kafka等相關技術博文，歡迎關注實時流式計算
