Kafka Shell基本命令（包括topic的增删改查）

时间 2019-12-14

标签 kafka shell 基本命令包括 topic 增删栏目 Kafka 繁體版

原文原文链接

转载请注明出处：http://www.cnblogs.com/xiaodf/java

建立kafka topic

bin/kafka-topics.sh --zookeeper node01:2181 --create --topic t_cdr --partitions 30  --replication-factor 2

注： partitions指定topic分区数，replication-factor指定topic每一个分区的副本数node

partitions分区数:
- partitions ：分区数，控制topic将分片成多少个log。能够显示指定，若是不指定则会使用broker(server.properties)中的num.partitions配置的数量
- 虽然增长分区数能够提供kafka集群的吞吐量、可是过多的分区数或者或是单台服务器上的分区数过多，会增长不可用及延迟的风险。由于多的分区数，意味着须要打开更多的文件句柄、增长点到点的延时、增长客户端的内存消耗。
- 分区数也限制了consumer的并行度，即限制了并行consumer消息的线程数不能大于分区数
- 分区数也限制了producer发送消息是指定的分区。如建立topic时分区设置为1，producer发送消息时经过自定义的分区方法指定分区为2或以上的数都会出错的；这种状况能够经过alter –partitions 来增长分区数。
replication-factor副本
- replication factor 控制消息保存在几个broker(服务器)上，通常状况下等于broker的个数。
- 若是没有在建立时显示指定或经过API向一个不存在的topic生产消息时会使用broker(server.properties)中的default.replication.factor配置的数量

查看全部topic列表

bin/kafka-topics.sh --zookeeper node01:2181 --list

查看指定topic信息

bin/kafka-topics.sh --zookeeper node01:2181 --describe --topic t_cdr

控制台向topic生产数据

bin/kafka-console-producer.sh --broker-list node86:9092 --topic t_cdr

控制台消费topic的数据

bin/kafka-console-consumer.sh  --zookeeper node01:2181  --topic t_cdr --from-beginning

查看topic某分区偏移量最大（小）值

bin/kafka-run-class.sh kafka.tools.GetOffsetShell --topic hive-mdatabase-hostsltable  --time -1 --broker-list node86:9092 --partitions 0

注： time为-1时表示最大值，time为-2时表示最小值shell

增长topic分区数

为topic t_cdr 增长10个分区服务器

bin/kafka-topics.sh --zookeeper node01:2181  --alter --topic t_cdr --partitions 10

删除topic，慎用，只会删除zookeeper中的元数据，消息文件须手动删除

bin/kafka-run-class.sh kafka.admin.DeleteTopicCommand --zookeeper node01:2181 --topic t_cdr

查看topic消费进度

这个会显示出consumer group的offset状况，必须参数为--group，不指定--topic，默认为全部topic微信

Displays the: Consumer Group, Topic, Partitions, Offset, logSize, Lag, Owner for the specified set of Topics and Consumer Group工具

bin/kafka-run-class.sh kafka.tools.ConsumerOffsetChecker

required argument: [group] 
Option Description 
------ ----------- 
--broker-info Print broker info 
--group Consumer group. 
--help Print this message. 
--topic Comma-separated list of consumer 
   topics (all topics if absent). 
--zkconnect ZooKeeper connect string. (default: localhost:2181)

Example,

bin/kafka-run-class.sh kafka.tools.ConsumerOffsetChecker --group pv

Group           Topic              Pid Offset   logSize    Lag    Owner 
pv              page_visits        0   21       21         0      none 
pv              page_visits        1   19       19         0      none 
pv              page_visits        2   20       20         0      none

以上图中参数含义解释以下：
topic：建立时topic名称
pid：分区编号
offset：表示该parition已经消费了多少条message
logSize：表示该partition已经写了多少条message
Lag：表示有多少条message没有被消费。
Owner：表示消费者大数据

细看kafka-run-class.sh脚本，它是调用了ConsumerOffsetChecker的main方法，因此，咱们也能够经过java代码来访问scala的ConsumerOffsetChecker类，代码以下：ui

import kafka.tools.ConsumerOffsetChecker;  
  
/** 
 * kafka自带不少工具类，其中ConsumerOffsetChecker能查看到消费者消费的状况, 
 * ConsumerOffsetChecker只是将信息打印到标准的输出流中 
 * 
 */  
public class RunClass  {  
    public static void main(String[] args)  {  
        //group-1是消费者的group名称,能够在zk中  
        String[] arr = new String[]{"--zookeeper=192.168.199.129:2181,192.168.199.130:2181,192.168.199.131:2181/kafka","--group=group-1"};  
        ConsumerOffsetChecker.main(arr);  
    }  
}

更多大数据技术干货，欢迎关注“大数据技术进阶”微信公众号
this