kafka 日常操作

创建 topic

/usr/local/kafka/bin/kafka-topics.sh --zookeeper localhost:2181 --replication-factor 3 --partitions 10 --create --topic test

查看 topic

/usr/local/kafka/bin/kafka-topics.sh --list --zookeeper localhost:2181 | grep test

查看指定 toopic 信息

/usr/local/kafka/bin/kafka-topics.sh --zookeeper localhost:2181 --describe --topic test

删除topic

/usr/local/kafka/bin/kafka-topics.sh --zookeeper localhost:2181 --delete --topic test

查看 topic 消费数据

/usr/local/kafka/bin/kafka-console-consumer.sh --bootstrap-server 10.4.22.39:9099 --topic test --from-beginning

生产 topic 数据

/usr/local/kafka/bin/kafka-console-producer.sh --broker-list 10.4.22.39:9092 --topic test  

增加 topic 分区数

/usr/local/kafka/bin/kafka-topics.sh --zookeeper localhost:2181 --alter --topic test --partitions 10

查看消费组

/usr/local/kafka/bin/kafka-consumer-groups.sh --bootstrap-server localhost:9099 --list

删除消费组

若配置文件中server.properties没有配置delete.topic.enable=true，那么此时的删除并不是真正的删除，而是把该topic标记为marked for deletion，不会释放磁盘空间。若想释放磁盘空间可以使用下面的方法。

/usr/local/kafka/bin/kafka-consumer-groups.sh --bootstrap-server 10.4.11.12:9092 --delete --group test

将topic限制为60G（60737418240bytes）

/usr/local/kafka/bin/kafka-configs.sh --zookeeper localhost:2181 --alter --entity-name topic_test_xxxxxx --entity-type topics --add-config  retention.bytes=60737418240

kafka 动态调整数据保留时间为3天(259200000ms)

kafka-configs.sh --zookeeper localhost:2181 --entity-type topics --entity-name test --alter --add-config retention.ms=259200000

查看消费组状态

watch -d /usr/local/kafka/bin/kafka-consumer-groups.sh --bootstrap-server 10.4.22.39:9099 --describe --group mygroupt

查看所有broker节点

#先连接到zookeeper
/usr/local/zookeeper/bin/zkCli.sh
[zk: localhost:2181(CONNECTED) 2] ls /
[admin, brokers, cluster, config, consumers, controller, controller_epoch, isr_change_notification, latest_producer_id_block, log_dir_event_notification, zookeeper]
[zk: localhost:2181(CONNECTED) 5] ls /brokers/ids
[1, 2, 3, 4, 5, 6, 7, 8, 9]

Kafka 动态调整副本数量

vim replication.json

{
    "version": 1, 
    "partitions": [
        {
            "topic": "dba_test",                   #topic
            "partition": 0, 				#分区数量
            "replicas": [				#新增副本数量  原有 121 新增 141、131
                121, 
                141, 
                131
            ]
        }
    ]
}

#执行新增命令
/usr/local/kafka/bin/kafka-reassign-partitions.sh --zookeeper localhost:2181 --reassignment-json-file replication.json --execute
# 查看topic详情
/usr/local/kafka/bin/kafka-topics.sh --zookeeper localhost:2181 --describe --topic dba_test