首页 > 其他 > 详细

kafka offset 设置

时间:2017-05-18 09:29:44      阅读:525      评论:0      收藏:0      [点我收藏+]
http://stackoverflow.com/questions/36579815/kafka-python-how-do-i-commit-a-partition
from kafka import KafkaConsumer
from kafka import TopicPartition from kafka.structs import OffsetAndMetadata ... topic = ‘your_topic‘ partition = 0 tp = TopicPartition(topic,partition) kafkaConsumer = KafkaConsumer(config here...) kafkaConsumer.assign([tp]) offset = 15394125 kafkaConsumer.commit({ tp: OffsetAndMetadata(offset, None) })



meta = consumer.partitions_for_topic(topic)
options = {}
options[partition] = OffsetAndMetadata(message.offset, meta)
consumer.commit(options)

如果consumer.commit()不可以,可以使用seek(),使用seek(),如果有多个partition,需要为每个partition都手动进行consumer assign.
topic_partition = TopicPartition("TOPIC_TEST", 1)
# 格式为topic, partition, 1表示partition 1.
consumer.assign([topic_partition])

consumer.seek(topic_partition, 1660000)
同样的代码,有些用commit可以,有些不可以,用seek()就可以。
 

kafka offset 设置

原文:http://www.cnblogs.com/buxizhizhoum/p/6871663.html

(0)
(0)
   
举报
评论 一句话评论(0
关于我们 - 联系我们 - 留言反馈 - 联系我们:wmxa8@hotmail.com
© 2014 bubuko.com 版权所有
打开技术之扣,分享程序人生!