kafka 常用操作指令

建立topic1

./kafka-topics.sh --create --zookeeper 192.168.1.93:6383 --replication-factor 1 --partitions 1 --topic patrol_listen

建立topic2

./kafka-topics.sh --create --zookeeper 192.168.1.93:6383 --replication-factor 1 --partitions 1 --topic patrol_result

生産消息到topic2

  ./kafka-console-producer.sh --broker-list localhost:9092 --topic patrol_result  

查詢topic2中的消息

  ./kafka-console-consumer.sh -zookeeper localhost:2181 --from-beginning --topic patrol_result 

#!/bin/sh

## 參數定義

dt=`date +"%Y%m%d" -d "-1 days"`

outpath=/xxxx_log_${dt}.txt    

brokerlist=192.168.1.100:9092,192.168.1.101:9092,192.168.1.102:9092

echo $dt $outpath $brokerlist

## 查詢hive表，輸出檔案到指定路徑

hive -e "select name,age from test" > ${outpath}

## 判斷檔案大小，如果大于0，則加載檔案，寫入kafka

## 注意 結尾的 | > out.txt要加上，否則會出現很多奇怪的大于号

fileSize=`du -b ${outpath} | awk '{print $1}'`

if [ $fileSize -gt 0 ] then

   cat ${outpath} | ./kafka_2.11-1.0.0/bin/kafka-console-producer.sh --broker-list ${brokerlist} --sync --topic test1 | > out.txt

fi

檢視有哪些主題：

./kafka-topics.sh --list --zookeeper localhost:2181

檢視topic的詳細資訊

./kafka-topics.sh -zookeeper 127.0.0.1:2181 -describe -topic testKJ1

為topic增加副本

./kafka-reassign-partitions.sh -zookeeper 127.0.0.1:2181 -reassignment-json-file json/partitions-to-move.json -execute

建立topic

./kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic testKJ1

為topic增加partition

./bin/kafka-topics.sh -zookeeper 127.0.0.1:2181 -alter -partitions 20 -topic testKJ1

kafka生産者用戶端指令

./kafka-console-producer.sh --broker-list localhost:9092 --topic testKJ1

kafka消費者用戶端指令

./kafka-console-consumer.sh -zookeeper localhost:2181 --from-beginning --topic testKJ1

kafka服務啟動

./kafka-server-start.sh ../config/server.properties

下線broker

./kafka-run-class.sh kafka.admin.ShutdownBroker --zookeeper 127.0.0.1:2181 --broker #brokerId# --num.retries 3 --retry.interval.ms 60

s

shutdown broker

删除topic

./kafka-run-class.sh kafka.admin.DeleteTopicCommand --topic testKJ1 --zookeeper 127.0.0.1:2181

./kafka-topics.sh --zookeeper localhost:2181 --delete --topic testKJ1

檢視consumer組内消費的offset

./kafka-run-class.sh kafka.tools.ConsumerOffsetChecker --zookeeper localhost:2181 --group test --topic testKJ1

./kafka-consumer-offset-checker.sh --zookeeper localhost:12181 --group group1 --topic group1

./kafka-consumer-groups.sh --bootstrap-server localhost:9092 --describe --group group1

設定為最初偏移量：

./kafka-consumer-groups.sh --bootstrap-server snn:6667 --group offsettest --topic offset-test --reset-offsets --to-earliest –execute

設定任意偏移量：

./kafka-consumer-groups.sh --bootstrap-server snn:6667 --group offsettest --topic offset-test --reset-offsets --to-offset 3 –execute

設定最近偏移量

./kafka-consumer-groups.sh --bootstrap-server snn:6667 --group offsettest --topic offset-test --reset-offsets --to-latest --execute