分享

storm中关于kafkSpout的问题求教

caiyifeng 发表于 2015-8-26 13:47:42 [显示全部楼层] 回帖奖励 阅读模式 关闭右栏 3 11393
我在storm平台发布了一个Topology(例如 testTopology),通过kafkaSpout 从kafka上抓取消息进行处理;
storm会在 zookeeper的节点 /storm/testTopology/partition_0  中记录当前处理的消息的offset,例如:
{"topology":{"id":"e20a3347-f43b-4765-94fc-2aec6b8a7d25","name":"testTopology"},"offset":4305106,"partition":0,"broker":{"host":"gjzq-zz5-20","port":9092},"topic":"test-topic"}
说明当前处理到的消息的offset 是4305106

我的问题是:
假设我想指定任意的offset,让Topology从该offset开始进行处理,应该怎么做??
我曾经将/storm/testTopology/partition_0 节点中的offset设置为一个新值,但没有任何作用,反而storm无法抓取到kafka的消息了


请大神解答。。;。多谢!!!

已有(3)人评论

跳转到指定楼层
Alkaloid0515 发表于 2015-8-26 14:44:10
应该参数的问题

KafkaConfig的startOffsetTime字段。默认为kafka.api.OffsetRequest.EarliestTime()开始读,也就是从Kafka中最早的消息开始处理。也可以设成kafka.api.OffsetRequest.LatestOffset,也就是最早的消息开始读。也可以自己指定具体的值。
回复

使用道具 举报

hyj 发表于 2015-8-26 17:56:32
提示: 作者被禁止或删除 内容自动屏蔽
回复

使用道具 举报

caiyifeng 发表于 2015-8-26 18:14:25
回复

使用道具 举报

您需要登录后才可以回帖 登录 | 立即注册

本版积分规则

关闭

推荐上一条 /2 下一条