当前位置:   article > 正文

kafka 命令行 生产消费数据,查看偏移量,修改偏移量,修改数据保留时间_kafka修改偏移量

kafka修改偏移量

kafka 命令行 创建topic 查看topic详情 生产消费数据,查看偏移量,修改分区偏移量(多方法),修改分区数量、修改数据保留时间

1.知识点

1)Topic相关:创建Topic、删除Topic、查看Topic列表、查看Topic详细信息

2)生产者相关:往某个Topic中生产数据

3)消费者相关:从某个Topic中消费数据

4)消费组(group)相关:查看消费者group、查看消费者消费情况(消费至那个offset/积压数据量多少)

5)修改topic下某个消费者(groupid)所有分区或某个分区的offset为任意指定偏移量

2.实现命令

以下命令若非特别说明均在kafka安装目录的bin目录下执行

1)Topic相关:创建Topic、删除Topic、查看Topic详细信息、查看Topic列表、修改topic分区数

TOP相关的命令执行连接zookeeper 端口2181

创建Topic

# 创建一个3分区1副本名为test的topic,必须指定分区数 --partitions 和副本数--replication-factor,其中副本数量不能超过kafka节点(broker)数量
./kafka-topics.sh --zookeeper localhost:2181  --topic test --partitions 3 --replication-factor 1 --create
  • 1
  • 2

删除Topic

# 删除名为test的topic
# 删除topic时只有在kafka安装目录config目录下的server.properties中将delete.topic.enable 设置为true topic才会真实删除,否则只是标记为删除,实则不会删除
./kafka-topics.sh --zookeeper localhost:2181  --topic test  --delete
  • 1
  • 2
  • 3

查看某个Topic 分区 副本信息

# 查看名为test的topic的详细信息,分区 副本的数量
./kafka-topics.sh --zookeeper localhost:2181  --topic test --describe
  • 1
  • 2

查看有那些Topic

# 查看kafka中创建了那些topic
./kafka-topics.sh  --zookeeper localhost:2181 --list
  • 1
  • 2

修改topic的分区数

# 将名为test的topic 修改为4个分区
# 注意 分区数只能增加不能减少
./kafka-topics.sh --zookeeper localhost:2181 -alter --partitions 4 --topic test
  • 1
  • 2
  • 3
2)生产者相关:往某个topic中生产数据

生产者相关命令执行连接broke-list 端口9092

使用命令行往某个topic中写入数据

# 使用命令行 给名为 test 的topic 中生产数据
# 执行以下命令,然后在命令行中写入要发送kafka的数据回车即可发送数据到kafka
./kafka-console-producer.sh --broker-list localhost:9092 --topic test
  • 1
  • 2
  • 3

如果执行上述命令报错:

在这里插入图片描述

则将命令中的localhost 更换为主机名即可。

3)消费者相关:从某个topic中消费数据

消费者相关可连接zookeeper 端口2181 或者bootstrap 端口9092, 0.8版本及以下版本kafka只能连接zookeeper,0.9版本及以上版本建议连接bootstrap ,但也可连接zookeeper

消费某个topic中的最新数据

# 0.8版本及以下的的kafka 使用如下命令test topic中的数据
./kafka-console-consumer.sh  --zookeeper localhost:2181 --topic test
# 指定消费10条数据
./kafka-console-consumer.sh  --zookeeper localhost:2181 --topic test --max-messages 10
# 0.9版本及以上的kafka建议使用如下命令进行消费,当然也可使用上一条命令消费
./kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6

消费某个topic中最老的数据

# 0.8版本及以下的的kafka 使用如下命令test topic中的数据
./kafka-console-consumer.sh  --zookeeper localhost:2181 --topic test --from-beginning
# 0.9版本及以上的kafka建议使用如下命令进行消费,当然也可使用上一条命令消费
./kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test --from-beginning
  • 1
  • 2
  • 3
  • 4

消费某个topic中的数据并指定groupid

# 在命令行消费某个topic中的数据通过/config/consumer.properties 配置文件指定groupid 
# 0.8版本及以下的的kafka 使用如下命令test topic中的数据
./kafka-console-consumer.sh --zookeeper localhost:2181 --topic test --consumer.config ../config/consumer.properties
# 0.9版本及以上的kafka建议使用如下命令进行消费,当然也可使用上一条命令消费
./kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test --consumer.config ../config/consumer.properties
  • 1
  • 2
  • 3
  • 4
  • 5

将/config/consumer.properties配置文件中groupid对应的offset删除,该groupid重置为未使用状态

# 使用这条命令会从最新消息开始消费,会将之前groupid记录的offset重置,并重新开始记录
./kafka-console-consumer.sh --zookeeper localhost:2181 --topic test --consumer.config ../config/consumer.properties  --delete-consumer-offsets
# 使用consumer.properties 不可以和--from-beginning一同使用 除非与--delete-consumer-offsets一同使用
# 使用这条命令会从头开始消费数据,会将之前groupid记录的offset重置,并重新开始记录
./kafka-console-consumer.sh --zookeeper localhost:2181 --topic test --consumer.config ../config/consumer.properties  --delete-consumer-offsets --from beginning
  • 1
  • 2
  • 3
  • 4
  • 5
4)消费组(group)相关:查看消费者group、查看消费者消费情况(消费至那个offset/积压数据量多少)

查看有那些消费者group

# 0.8版本及以下的的kafka 使用如下命令查看有那些消费者group
./kafka-consumer-groups.sh  --zookeeper localhost:2181 --list
# 0.9版本及以上的kafka建议使用如下命令查看有那些消费者group,当然也可使用上一条命令消费
./kafka-consumer-groups.sh --bootstrap-server localhost:9092 --list
  • 1
  • 2
  • 3
  • 4

查看某个消费者消费情况(消息队列堆积情况)

# 0.8版本及以下的的kafka 使用如下命令查看名为testgroup 的消费组的消费情况
./kafka-consumer-groups.sh --zookeeper localhost:2181 --group testgroup --describe
# 0.9版本及以上的的kafka 使用如下命令查看名为testgroup 的消费组的消费情况
./kafka-consumer-groups.sh --bootstrap-server localhost:9092 --group testgroup --describe
  • 1
  • 2
  • 3
  • 4
5)修改某个消费组的偏移量(offset)

通过zk客户端对topic的分区修改offset 为任意偏移量

# 独立安装的zk,进入zookeeper安装目录的bin目录下,使用如下命令进入zk客户端
./zkCli.sh -server localhost:2181
# 非独立安装的的zk, 直接在kafka安装目录bin目录下,使用如下命令进入zk客户端
./zookeeper-shell.sh  localhost:2181
# 进入zk客户端后可查看某个分区的偏移量 例如名为test的topic的消费者组 test-consumer-group 0分区的offset的消费情况
get /consumers/test-consumer-group/offsets/test/0
# 设置名为test的topic的消费者组 test-consumer-group 0分区的offset 为1000
set /consumers/test-consumer-group/offsets/test/0  1000
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8

通过kafka内置的kafka.tools.UpdateOffsetsInZK类实现修改某个topic 的消费组(config/consumer.properties中配置的groupid)的所有分区的偏移量为最新(latest)或者最旧(earliest)

# 将名为test的topic的消费组(groupid必须从consumer.properties获取,即需要将需要修改的groupid写入consumer.properties配置文件)所有分区的offset设置为最早earliest
./kafka-run-class.sh kafka.tools.UpdateOffsetsInZK earliest ../config/consumer.properties test
# 将名为test的topic的消费组(groupid必须从consumer.properties获取,即需要将需要修改的groupid写入consumer.properties配置文件)所有分区的offset设置为最新latest
./kafka-run-class.sh kafka.tools.UpdateOffsetsInZK latest ../config/consumer.properties test
  • 1
  • 2
  • 3
  • 4

0.11.0.0及以上版本修改偏移量可使用Kafka自带的kafka-consumer-groups.sh脚本

# 以下可将--zookeeper localhost:2181 更换为--bootstrap-server localhost:9092 高版本的消费者建议连接bootstrap
# 将test topic的消费组test-consumer-group的0分区的偏移量设置为最新
./kafka-consumer-groups.sh --zookeeper localhost:2181 --group test-consumer-group --topic test:0 --reset-offsets --to-earliest –execute
# 将test topic的消费组test-consumer-group的0和1分区的偏移量设置为最旧
./kafka-consumer-groups.sh --zookeeper localhost:2181 --group test-consumer-group --topic test:0,1 --reset-offsets --to-latest –execute
# 将test topic的消费组test-consumer-group的所有分区的偏移量设置为1000
./kafka-consumer-groups.sh --zookeeper localhost:2181 --group test-consumer-group --topic test --reset-offsets --to-offset 1000 –execute
# --reset-offsets后可以跟的其他用法:--to-current:把位移调整到分区当前位移
# --reset-offsets后可以跟的其他用法:--shift-by N: 把位移调整到当前位移 + N处,注意N可以是负数,表示向前移动
# --reset-offsets后可以跟的其他用法:--to-datetime <datetime>:把位移调整到大于给定时间的最早位移处,datetime格式是yyyy-MM-ddTHH:mm:ss.xxx,比如2017-08-04T00:00:00.000
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10

6)修改 Kafka 主题(topic)的消息保留时间(retention.ms)

kafka-topics.sh 用于管理 Kafka 的主题(topics)
# 查看 Kafka 中指定主题(topic_name)的详细信息,包括分区数量、副本分布以及配置信息等
./kafka-topics.sh --zookeeper localhost:2181 --describe --topic topic_name
# 更改名为 topic_name 的主题的 retention.ms 属性,将消息保留时间设置为12小时,这个命令会立即生效,修改后新写入的消息将按照新的保留时间策略进行处理,但是已经存在于主题中的消息不受影响。
./kafka-topics.sh --zookeeper localhost:2181 --alter --topic topic_name --config retention.ms=43200000 
# 删除名为 topic_name 的主题的 retention.ms 属性,从而取消该属性的配置。这将导致该主题的消息保留时间采用默认值
./kafka-topics.sh --zookeeper localhost:2181 --alter --topic topic_name --delete-config retention.ms 

kafka-config.sh 用于管理 Kafka 的配置
# 查看 Topic 的配置
./kafka-config.sh --zookeeper localhost:2181 --describe --entity-type topics --entity-name topic_name 
# 修改 Topic 的配置 修改 topic 消息的保存为12时间,超过该时间,消息将被删除,这个命令需要等待 ZooKeeper 中的数据同步到 Kafka 服务器中才能生效,修改后新写入的消息将按照新的保留时间策略进行处理,而且已经存在于主题中的消息也会按照新的保留时间策略进行处理。这个命令可以对多个主题进行配置修改,而 kafka-topics.sh 命令只能修改单个主题的配置。
./kafka-config.sh --zookeeper localhost:2181 --alter --entity-type topics --entity-name topic_name --add-config retention.ms=43200000

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
声明:本文内容由网友自发贡献,不代表【wpsshop博客】立场,版权归原作者所有,本站不承担相应法律责任。如您发现有侵权的内容,请联系我们。转载请注明出处:https://www.wpsshop.cn/w/菜鸟追梦旅行/article/detail/621214
推荐阅读
相关标签
  

闽ICP备14008679号