Spark流Kafka偏移管理

pftdvrlh 于 2021-06-07 发布在 Kafka

关注(0)|答案(1)|浏览(329)

我一直在做Spark流的工作，消费者和生产数据通过Kafka。我用的是directdstream，所以我必须自己管理偏移量，我们采用redis来写和读偏移量，现在有一个问题，我启动客户端时，客户端需要从redis得到偏移量，而不是kafka本身的偏移量，我怎么写代码呢？现在我已经写了下面的代码：

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))

但是我认为fromOffset是spark流媒体客户端启动时的值（来自redis），而不是在运行时的值。

apache-kafka apache-spark spark-streaming spark-streaming-kafka Offset

来源：https://stackoverflow.com/questions/49798501/spark-streaming-kafka-offset-manage

1条答案

按热度按时间

dohp0rv51#

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

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)

赞(0）回复(0）举报 2021-06-07

我来回答

Spark流Kafka偏移管理

1条答案

相关问题

热门标签

最新问答