Kafka-shell常用操作

墨蓝 2021-09-23 06:42 268阅读 0赞

> **kafka 的shell操作(基于topic的增删改查 shell)**

**目录**

*  1. 启动节点的Kafka进程
 *  2. 创建topic
 *  3. 查看已创建的所有topics
 *  4. 查看指定topic详细信息
 *  5. 开启生产者模式
 *  6. 开启消费者模式
 *  7. 查看分区偏移量
 *  8. 增加topic分区数
 *  9. 删除topic
 *  10. 查看队列信息
 *  11. 查看指定队列消费数据

--------------------

### 1. 启动节点的Kafka进程 ###

CDH已经启动可以忽略

nohup kafka-server-start.sh \
    /opt/cloudera/parcels/CDH-6.2.0-1.cdh6.2.0.p0.967373/lib/kafka/config/server.properties \
    
    1>~/logs/kafka_std.log \
    2>~/logs/kafka_err.log &

### 2. 创建topic ###

kafka-topics.sh \
    --create \
    --zookeeper cdh-master:2181,cdh-slave01:2181,cdh-slave02:2181 \
    --partitions 1 \
    --replication-factor 1 \
    --topic kafka_test
    
    kafka-topics.sh \
    --create \
    --zookeeper cdh-master:2181,cdh-slave01:2181,cdh-slave02:2181 \
    --replication-factor 3 \
    --partitions 10 \
    --topic kafka_test

### 3. 查看已创建的所有topics ###

kafka-topics.sh \
    --zookeeper cdh-master:2181,cdh-slave01:2181,cdh-slave02:2181 \
    --list

### 4. 查看指定topic详细信息 ###

kafka-topics.sh \
    --zookeeper cdh-master:2181,cdh-slave01:2181,cdh-slave02:2181 \
    --describe \
    --topic kafka_test

### 5. 开启生产者模式 ###

kafka-console-producer.sh \
    --broker-list cdh-master:9092,cdh-slave01:9092,cdh-slave02:9092 \
    --topic kafka_test

### 6. 开启消费者模式 ###

kafka-console-consumer.sh \
    --bootstrap-server cdh-master:9092,cdh-slave01:9092,cdh-slave02:9092 \
    --from-beginning \
    --topic kafka_test

### 7. 查看分区偏移量 ###

kafka-run-class.sh \        # CDH6为：kafka-run-class
    kafka.tools.GetOffsetShell \
    --broker-list cdh-master:9092,cdh-slave01:9092,cdh-slave02:9092 \
    --topic topic_sdk_log \
    --partitions 0 \
    --time -1

### 8. 增加topic分区数 ###

kafka-topics.sh \
    --zookeeper cdh-master:2181,cdh-slave01:2181,cdh-slave02:2181 \
    --alter \
    --topic kafka_test \
    --partitions 20  
    
    # 增加副本数
    kafka-topics.sh \
    --zookeeper cdh-master:2181,cdh-slave01:2181,cdh-slave02:2181 \
    --alter \
    --topic kafka_test \
    --replication-factor 2

### 9. 删除topic ###

kafka-topics.sh \
    --zookeeper cdh-master:2181,cdh-slave01:2181,cdh-slave02:2181 \
    --delete \
    --topic kafka_test

### 10. 查看队列信息 ###

kafka-consumer-groups.sh \
    --bootstrap-server cdh-slave01:9092 \
    --list

### 11. 查看指定队列消费数据 ###

新版本，consumer 信息存储在 Kafka 的 broker 中，Kafka自维护偏移量，。

kafka-consumer-groups.sh \
    --bootstrap-server cdh-slave01:9092 \
    --describe \
    --group flume

旧版本，consumer 信息存储在 Zookeeper 中，Zookeeper 维护偏移量。

kafka-consumer-groups.sh \
    ----zookeeper cdh-slave01:2181 \
    --describe \
    --group flume

# 进入zk，多节点逗号分隔
    zkCli.sh -server hadoop-master:2181
    
    # 查看某个group_id的某个topic的offset
    ls /consumers/[group_id]/offsets/[topic]/[broker_id-part_id]