創(chuàng)建主題径荔，replication?分區(qū)?partitions?副本數(shù)，對(duì)應(yīng)broker數(shù)量
kafka-topics.sh --create --zookeeper ${host}:2181 --replication-factor 1 --partitions 1 --topic save-data
查看主題列表
kafka-topics.sh --list --zookeeper ${host}:2181
查看特定主題狀態(tài)
kafka-topics.sh --zookeeper ${host}:2181 --topic save-data --describe
刪除主題
kafka-topics.sh --zookeeper ${host}:2181 --topic save-data2 --delete
查看消費(fèi)組列表
kafka-consumer-groups.sh --bootstrap-server ${host}:9092 --list
查看特定消費(fèi)組狀態(tài)
kafka-consumer-groups.sh --bootstrap-server ${host}:9092 --group save-data2-group-0 --describe
設(shè)置消費(fèi)組id
kafka-console-consumer.sh --bootstrap-server ${host}:9092 --topic save-data3 --consumer-property group.id=save-data3-group-0
重設(shè)消費(fèi)組偏移量
kafka-consumer-groups.sh --bootstrap-server ${host}:9092 --group save-data2-group-0 --reset-offsets --all-topics --to-earliest --execute
創(chuàng)建新消費(fèi)組
kafka-console-consumer.sh --bootstrap-server ${host}:9092 --topic save-data --from-beginning --new-consumer
kafka-console-producer.sh --topic save-data --bootstrap-server ${host}:9092
1、集群管理
前臺(tái)啟動(dòng)broker
bin/kafka-server-start.sh <path>/server.properties
Ctrl + C 關(guān)閉
后臺(tái)啟動(dòng)broker
bin/kafka-server-start.sh -daemon <path>/server.properties
關(guān)閉broker
bin/kafka-server-stop.sh
2、Topic管理
創(chuàng)建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管理
查詢消費(fèi)者組
bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --list
查詢消費(fèi)者組詳情
bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --describe --group groupname
重設(shè)消費(fèi)者組位移
最早處
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
某個(gè)位置
bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --group groupname --reset-offsets --all-topics --to-offset 2000 --execute
調(diào)整到某個(gè)時(shí)間之后得最早位移
bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --group groupname --reset-offsets --all-topics --to-datetime 2019-09-15T00:00:00.000
刪除消費(fèi)者組
bin/kafka-consumer-groups.sh --zookeeper localhost:2181 --delete --group groupname
4、腳本工具
producer腳本
bin/kafka-console-producer.sh --broker-list localhost:9092 --topic topicname
參數(shù)含義：
--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
指定分區(qū)
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當(dāng)前消息數(shù)
kafka-run-class.sh kafka.tools.GetOffsetShell --broker-list localhost:9092 --topic topicname --time -1
--time -1表示最大位移? --time -2表示最早位移
查詢consumeroffsets
bin/kafka-simple-consumer-shell.sh --topic _consumer_offsets --partition 12 --broker-list localhost:9092 --formatter "kafka.coorfinator.GroupMetadataManager\$OffsetsMessageFormatter"
5屹逛、MirrorMaker
跨機(jī)房災(zāi)備工具
bin/kafka-mirror-maker.sh --consumer.config consumer.properties --producer.config producer.properties --whitelist topicA|topicB
------------------------------------------------------
重新設(shè)置主題
示例move.json
{
? ? "version": 1,
? ? "partitions": [{
? ? ? ? ? ? "topic": "save-data2",
? ? ? ? ? ? "partition": 0,
? ? ? ? ? ? "replicas": [1003, 1004]
? ? ? ? },
? ? ? ? {
? ? ? ? ? ? "topic": "save-data2",
? ? ? ? ? ? "partition": 1,
? ? ? ? ? ? "replicas": [1004, 1003]
? ? ? ? }
? ? ]
}
kafka-reassign-partitions.sh --zookeeper ${host}:2181 --reassignment-json-file move.json --execute
修改主題分區(qū)和副本數(shù)
kafka-topics.sh --zookeeper ${host}:2181 --alter --partitions 2 --topic save-data2
觸發(fā)對(duì)所有的topic Leader進(jìn)行負(fù)載均衡
kafka-preferred-replica-election.sh --zookeeper ${host}:2181 --path-to-json-file move.json
重新選舉主題leader
kafka-leader-election.sh? --path-to-json-file move.json --bootstrap-server ${host}:9092 --election-type unclean
