Spark Streaming Kafka偏移量管理 [英] Spark Streaming kafka offset manage

问题描述

我一直在从事火花流工作，这些工作通过kafka进行消费并产生数据.我使用directDstream，所以我必须自己管理偏移量，我们使用redis来读写偏移量.现在有一个问题，当我启动客户端时，我的客户端需要从redis获取偏移量，而不是存在于kafka中的偏移量本身.如何显示我编写我的代码?现在我已在下面编写我的代码:

I had been doing spark streaming jobs which consumer and produce data through kafka. I used directDstream,so I had to manage offset by myself,we adopted redis to write and read offsets.Now there is one problem，when I launched my client,my client need to get the offset from redis,not offset which exists in kafka itself.how show I write my code?Now I had written my code below:

   kafka_stream = KafkaUtils.createDirectStream(
    ssc,
    topics=[config.CONSUME_TOPIC, ],
    kafkaParams={"bootstrap.servers": config.CONSUME_BROKERS,
                 "auto.offset.reset": "largest"},
    fromOffsets=read_offset_range(config.OFFSET_KEY))

但是我认为fromOffsets是启动流式传输客户端时(而不是在其运行期间)的值(来自redis).感谢您的帮助.

But I think the fromOffsets is the value(from redis) when the spark-streaming client lauched,not during its running.thank you for helpinp.

推荐答案

如果我对您的理解正确，则需要手动设置偏移量.这是我的方法:

If I understand you correctly you need to set your offset manually. This is how I do it:

from pyspark.streaming import StreamingContext
from pyspark.streaming.kafka import KafkaUtils
from pyspark.streaming.kafka import TopicAndPartition

stream = StreamingContext(sc, 120) # 120 second window

kafkaParams = {"metadata.broker.list":"1:667,2:6667,3:6667"}
kafkaParams["auto.offset.reset"] = "smallest"
kafkaParams["enable.auto.commit"] = "false"

topic = "xyz"
topicPartion = TopicAndPartition(topic, 0)
fromOffset = {topicPartion: long(PUT NUMERIC OFFSET HERE)}

kafka_stream = KafkaUtils.createDirectStream(stream, [topic], kafkaParams, fromOffsets = fromOffset)

这篇关于Spark Streaming Kafka偏移量管理的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！