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