kafka 指定offset怎么获得最后一条消息的offset

来源：蜘蛛抓取(WebSpider) 时间：2016-09-04 07:40 标签： 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范围为:[]
设置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开始读数据了。&
阅读(...) 评论()<!DOCTYPE HTML PUBLIC "-//W3C//DTD HTML 4.01 Transitional//EN"
您的访问请求被拒绝 403 Forbidden - ITeye技术社区
您的访问请求被拒绝
亲爱的会员，您的IP地址所在网段被ITeye拒绝服务，这可能是以下两种情况导致：
一、您所在的网段内有网络爬虫大量抓取ITeye网页，为保证其他人流畅的访问ITeye，该网段被ITeye拒绝
二、您通过某个代理服务器访问ITeye网站，该代理服务器被网络爬虫利用，大量抓取ITeye网页
请您点击按钮解除封锁&<!DOCTYPE HTML PUBLIC "-//W3C//DTD HTML 4.01 Transitional//EN"
您的访问请求被拒绝 403 Forbidden - ITeye技术社区
您的访问请求被拒绝
亲爱的会员，您的IP地址所在网段被ITeye拒绝服务，这可能是以下两种情况导致：
一、您所在的网段内有网络爬虫大量抓取ITeye网页，为保证其他人流畅的访问ITeye，该网段被ITeye拒绝
二、您通过某个代理服务器访问ITeye网站，该代理服务器被网络爬虫利用，大量抓取ITeye网页
请您点击按钮解除封锁&kafka怎么获得最后一条消息的offset_百度知道

我要回帖

更多关于 kafka 指定offset 的文章

 

随机推荐