Kafka常用topic操作命令汇总
topic 工具
https://cwiki.apache.org/confluence/display/KAFKA/Replication+tools
offset相关
# 最大offset
bin/kafka-run-class.sh kafka.tools.GetOffsetShell --broker-list 192.168.32.43:9092 --topic test_topic --time -1
# 最小offset
bin/kafka-run-class.sh kafka.tools.GetOffsetShell --broker-list 192.168.32.43:9092 --topic test_topic --time -2
# offset
bin/kafka-run-class.sh kafka.tools.GetOffsetShell --broker-list 192.168.32.43:9092 --topic test_topic
topic相关
# 创建topic
bin/kafka-topics.sh --create --zookeeper localhost:2181 --topic test_topic --replication-factor 3 --partitions 1
bin/kafka-topics.sh --create --zookeeper localhost:2181 --topic test_topic --replication-factor 3 --partitions 50 --config cleanup.policy=compact
bin/kafka-topics.sh --zookeeper localhost:2181 --create --topic my-topic --partitions 1 --replication-factor 1 --config max.message.bytes=64000 --config flush.messages=1
# 查看topic
bin/kafka-topics.sh --zookeeper localhost:2181 --describe --topic test_topic
# 修改topic
bin/kafka-configs.sh --alter --zookeeper localhost:2181 --entity-name test_topic --entity-type topics --add-config cleanup.policy=compact
bin/kafka-configs.sh --alter --zookeeper localhost:2181 --entity-name test_topic --entity-type topics --delete-config cleanup.policy
consumer-group相关
bin/kafka-consumer-groups.sh --bootstrap-server 192.168.21.11:9092,192.168.21.12:9092,192.168.21.13:9092 --group test_group --describe
bin/kafka-consumer-groups.sh --new-consumer --bootstrap-server localhost:9092 --list
bin/kafka-consumer-groups.sh --zookeeper 127.0.0.1:2181 --group test_group --describe
consumer相关
bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test_topic
# 带有key
bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test_topic --property print.key=true
bin/kafka-console-consumer.sh --zookeeper localhost:2181 --topic test_topic
producer相关
bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test_topic
bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test_topic --property "parse.key=true" --property "key.separator=:"
producer-golang
# golang实现的kafka客户端
https://github.com/Shopify/sarama/tree/master/tools
# Minimum invocation
kafka-console-producer -topic=test -value=value -brokers=kafka1:9092
# It will pick up a KAFKA_PEERS environment variable
export KAFKA_PEERS=kafka1:9092,kafka2:9092,kafka3:9092
kafka-console-producer -topic=test -value=value
# It will read the value from stdin by using pipes
echo "hello world" | kafka-console-producer -topic=test
# Specify a key:
echo "hello world" | kafka-console-producer -topic=test -key=key
# Partitioning: by default, kafka-console-producer will partition as follows:
# - manual partitioning if a -partition is provided
# - hash partitioning by key if a -key is provided
# - random partioning otherwise.
#
# You can override this using the -partitioner argument:
echo "hello world" | kafka-console-producer -topic=test -key=key -partitioner=random
# Display all command line options
kafka-console-producer -help
