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 等相关技术博文, 欢迎关注实时流式计算
来源: https://www.cnblogs.com/tree1123/p/11525610.html