电火花流卡夫卡偏移管理

Question

我一直在做星火流工作，这些工作是通过卡夫卡( kafka )来消费和生产数据的。我使用了directDstream，所以我不得不自己管理偏移量，我们用redis编写和读取offsets.Now有一个问题，当我启动我的客户端时，我的客户需要从redis获得偏移量，而不是使用itself.how中存在的偏移量来显示我编写的代码?现在我已经编写了下面的代码：

   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)，而不是在它的running.thank you提供帮助的时候。

Answer 1

如果我正确地理解你，你需要手动设置你的偏移量。我就是这样做的：

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)