kafka 的offset的重置

最近在spark读取kafka消息时，每次读取都会从kafka最新的offset读取。但是如果数据丢失，如果在使用Kafka来分发消息，在数据处理的过程中可能会出现处理程序出异常或者是其它的错误，会造成数据丢失或不一致。这个时候你也许会想要通过kafka把数据从新处理一遍，或者指定kafka的offset读取。kafka默认会在磁盘上保存到7天的数据，你只需要把kafka的某个topic的consumer的offset设置为某个值或者是最小值，就可以使该consumer从你设置的那个点开始消费。这就需要从zk里面修改offset的值。

查询topic的offset的范围

用下面命令可以查询到topic:DynamicRange broker:SparkMaster:9092的offset的最小值：

$ /kafka/bin/kafka-run-class.sh kafka.tools.GetOffsetShell --broker-list SparkMaster:9092 —topic DynamicRange --time -2

输出

DynamicRange:0:1288

查询offset的最大值：

$ /kafka/bin/kafka-run-class.sh kafka.tools.GetOffsetShell --broker-list SparkMaster:9092 —topic DynamicRange --time -1

输出

DynamicRange:0:7885

从上面的输出可以看出topic:DynamicRange只有一个partition:0 offset范围为:[1288,7885]

设置consumer group的offset

启动zookeeper client

$ /opt/cloudera/parcels/CDH/lib/zookeeper/bin/zkCli.sh

通过下面命令设置consumer group:DynamicRangeGroup topic:DynamicRange partition:0的offset为1288:

set /consumers/DynamicRangeGroup/offsets/DynamicRange/0 1288

注意如果你的kafka设置了zookeeper root，比如为/kafka，那么命令应该改为：

set /kafka/consumers/DynamicRangeGroup/offsets/DynamicRange/0 1288

生效

重启相关的应用程序，就可以从设置的offset开始读数据了。

参考：https://metabroadcast.com/blog/resetting-kafka-offsets

时间： 2024-08-14 05:10:59

kafka 的offset的重置的相关文章

kafka的offset相关知识

Offset存储模型 由于一个partition只能固定的交给一个消费者组中的一个消费者消费,因此Kafka保存offset时并不直接为每个消费者保存,而是以 groupid-topic-partition -> offset 的方式保存. 如图所示: Kafka在保存Offset的时候,实际上是将Consumer Group和partition对应的offset以消息的方式保存在__consumers_offsets这个topic中. __consumers_offsets默认拥有50个par

kafka 消费者offset记录位置和方式

我们大家都知道,kafka消费者在会保存其消费的进度,也就是offset,存储的位置根据选用的kafka api不同而不同. 首先来说说消费者如果是根据javaapi来消费,也就是[kafka.javaapi.consumer.ConsumerConnector],我们会配置参数[zookeeper.connect]来消费.这种情况下,消费者的offset会更新到zookeeper的[consumers/{group}/offsets/{topic}/{partition}]目录下,例如: [z

kafka 相关命令 偏移重置

kafka官方文档 https://kafka.apache.org/documentation.html#quickstart kafka 安装文档 https://www.jianshu.com/p/c74e0ec577b0 相关命令: kafka 启动:kafka官方文档 https://kafka.apache.org/documentation.html#quickstart kafka 安装文档 https://www.jianshu.com/p/c74e0ec577b0 主题相关命

kafka将消费进度重置到最新的位置

/** *重置kafka消费进度 *参数中需要指定kafka集群中一台broker地址,要重置的topic名称,消费组,以及partition个数 */public static void seekLatest(String broker, String topic, String group, int partitionCnt){ Map<String, Object> configProps = new HashMap<>(); configProps.put(Consumer

关于kafka更改消费者对应分组下的offset值

kafka的offset保存位置分为两种情况 0.9.0.0版本之前默认保存在zookeeper当中 0.9.0.0版本之后保存在broker对应的topic当中 1.如何辨别你启用的consumer的offset保存位置进入zookeeper的命令行当中 zkCli.sh localhost:2181 用 ls / 查看目录如果你在代码中定义的group id 没有在 /consumers 这个文件夹中,代表offset保存在broker的topic中前提是consumer确实已经创建并启动如

从外部重置一个运行中consumer group的消费进度

对于0.10.1以上版本的kafka, 如何从外部重置一个运行中的consumer group的进度呢?比如有一个控制台,可以主动重置任意消费组的消费进度重置到12小时之前. 需要这么几个步骤: 1. 加入这个group 2. 踢掉所有其它group memeber 3. try assign all TopicPartition to this client 4. commit offsets 5. leave group 其中第二步是为了让自己当上leader,当然有可能不需要踢掉其它所有成

【转】Kafka某topic无法消费解决方案&amp;Kafka某Topic数据清理

由于项目原因,最近经常碰到Kafka消息队列某topic在集群宕机重启后无法消费的情况.碰到这种情况,有三步去判断原因所在: step A:如果用kafka串口(即console-consumer)是可以正常消费该topic,则排除kafka集群出现故障 step B:若平台业务能正常消费其他topic的消息,则排除平台业务代码逻辑问题 step C:不到万不得已,则只能手动删除kafka的对应topic的Log,但是清理Kafka Log又不能单纯的去删除中间环节产生的日志,中间关联的很多东西

kafka系列 -- 基础概念

kafka是一个分布式的.分区化.可复制提交的发布订阅消息系统 传统的消息传递方法包括两种: 排队:在队列中,一组用户可以从服务器中读取消息,每条消息都发送给其中一个人. 发布-订阅:在这个模型中,消息被广播给所有的用户. kafka与传统的消息传递技术相比优势之处在于: 快速:单一的Kafka代理可以处理成千上万的客户端,每秒处理数兆字节的读写操作. 可伸缩:在一组机器上对数据进行分区和简化,以支持更大的数据 持久:消息是持久性的,并在集群中进行复制,以防止数据丢失. 设计:它提供了容错保证和

kafka实战 - 删除topic

概述 在平时对kafka的运维工作中,我们经常会由于某些原因去删除一个topic,比如这个topic是测试用的,生产环境中需要删除.或者我想扩容topic的同时,这个topic中的数据我不想要了,这时候删除topic,增加broker,再重新创建topic就会是比较简单的方法.但是kafka删除topic时,有很多关键的点必须清楚,否则在删除topic的时候就会出现各种各样的问题. 我测试环境使用的kafka版本是0.10.2.0,不同版本的kafka默认配置和bin目录下脚本使用的方式略有不同