topic
1. 查看topic
选项说明：
- --list ：查看kafka所有的topic
- --bootstrap-server : 连接kafka集群
- --hadoop102:9092：hadoop102是指连接kafka任意一台机器，9092：kafka内部通信的端口
kafka-topics.sh --bootstrap-server hadoop102:9092 --list
2. 创建topic
选项说明：
- --topic : 定义topic名字
- --partitions : 定义分区数
- --replication-factor : 定义副本数
kafka-topics.sh --bootstrap-server hadoop102:9092 --create --topic first
--partitions 2 --replication-factor 2
3. 查看某个Topic的详情
选项说明：
- --topic first ： 查看指定的话题，如果不加此选项，则表示查看所有的话题
kafka-topics.sh --bootstrap-server hadoop102:9092 --describe --topic first
4.修改分区数
说明：
- --分区数只能增加不能减少 ，要不然消减的分区数据不好处理
- 分区内部消息有序，分区之间消息无序
kafka-topics.sh --bootstrap-server hadoop102:9092 --alter --topic first --partitions 6
5. 发送消息
选项说明：
- hadoop102:9092,hadoop103:9092,hadoop104:9092 : kafka的集群中的broker，
其实写一个也是可以的，写3个的目的只是避免当连接的kafka集群broker故障时连不上kafka集群的情况。
kafka-console-producer.sh --broker-list hadoop102:9092,hadoop103:9092,hadoop104:9092
--topic first
6. 消费消息
选项说明：
- --from-beginning ：
加上：会把topic中以往所有的数据都读取出来
不加：此时只会消费最新的数据，原来topic中的数据不会被消费
kafka-console-consumer.sh --bootstrap-server hadoop102:9092 --from-beginning --topic first
7. 删除topic
kafka-topics.sh --bootstrap-server hadoop102:9820 --delete --topic first
8. 查看指定消费者组的消费topic 的位置情况
kafka-consumer-groups.sh --bootstrap-server hadoop102:9092 --group test --describe
9. 设置指定消费者组的消费某个Topic 的偏移量
kafka-consumer-groups.sh --bootstrap-server hadoop102:9092 --group test --topic lkp_test --reset-offsets --to-offset 1 --execute
##指定主题、分区、消费偏移
kafka-consumer-groups.sh --bootstrap-server 10.202.13.27:9092 --group cjw --reset-offsets --topic cjw-test:1 --to-offset 2 --execute
10. 查询topic的offset的范围
用下面命令可以查询到topic:test broker:suna:9092的offset的最小值：
bin/kafka-run-class.sh kafka.tools.GetOffsetShell --broker-list suna:9092 -topic test --time -2
输出
test:0:1288
查询offset的最大值：
bin/kafka-run-class.sh kafka.tools.GetOffsetShell --broker-list hadoop102:9092 -topic test --time -1
输出
test:0:7885
从上面的输出可以看出topic:test只有一个partition:0 offset范围为:[1288,7885]
===========================================
Kafka Version为0.11.0.0之后，Consumer的Offset信息不再默认保存在Zookeeper上，而是选择用Topic的形式保存下来。
在命令行中可以使用kafka-consumer-groups的脚本实现Offset的相关操作。
更新Offset由三个维度决定：Topic的作用域、重置策略、执行方案。
## Topic的作用域
–all-topics：为consumer group下所有topic的所有分区调整位移）
–topic t1 --topic t2：为指定的若干个topic的所有分区调整位移
–topic t1:0,1,2：为指定的topic分区调整位移
## 重置策略
–to-earliest：把位移调整到分区当前最小位移
–to-latest：把位移调整到分区当前最新位移
–to-current：把位移调整到分区当前位移
–to-offset <offset>： 把位移调整到指定位移处
–shift-by N： 把位移调整到当前位移 + N处，注意N可以是负数，表示向前移动
–to-datetime <datetime>：把位移调整到大于给定时间的最早位移处，datetime格式是yyyy-MM-ddTHH:mm:ss.xxx，比如2017-08-04T00:00:00.000
–by-duration <duration>：把位移调整到距离当前时间指定间隔的位移处，duration格式是PnDTnHnMnS，比如PT0H5M0S
–from-file <file>：从CSV文件中读取调整策略
## 确定执行方案
什么参数都不加：只是打印出位移调整方案，不具体执行
–execute：执行真正的位移调整
–export：把位移调整方案按照CSV格式打印，方便用户成csv文件，供后续直接使用
## 注意事项
consumer group状态必须是inactive的，即不能是处于正在工作中的状态
不加执行方案，默认是只做打印操作
## 常用示例
更新到当前group最初的offset位置
bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --group test-group --reset-offsets --all-topics --to-earliest --execute
更新到指定的offset位置
bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --group test-group --reset-offsets --all-topics --to-offset 500000 --execute
更新到当前offset位置（解决offset的异常）
bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --group test-group --reset-offsets --all-topics --to-current --execute
offset位置按设置的值进行位移
bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --group test-group --reset-offsets --all-topics --shift-by -100000 --execute
offset设置到指定时刻开始
bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --group test-group --reset-offsets --all-topics --to-datetime 2017-08-04T14:30:00.000
=======================================================
网友评论