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

轉載請註明出處：http://www.cnblogs.com/xiaodf/

• 建立kafka topic
• 查看全部topic列表
• 查看指定topic信息
• 控制檯向topic生產數據
• 控制檯消費topic的數據
• 查看topic某分區偏移量最大（小）值
• 增長topic分區數
• 刪除topic，慎用，只會刪除zookeeper中的元數據，消息文件須手動刪除
• 查看topic消費進度

建立kafka topic

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

注： partitions指定topic分區數，replication-factor指定topic每一個分區的副本數

• 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時表示最小值

增長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類，代碼以下：

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);  
    }  
}

更多大數據技術乾貨，歡迎關注「大數據技術進階」微信公衆號