0

星火版本 = 2.3.0

卡夫卡版本 = 1.0.0

正在使用的代码片段：

# Kafka Enpoints
zkQuorum = '192.168.2.10:2181,192.168.2.12:2181' 
topic = 'Test_topic'

# Create a kafka Stream
kafkaStream = KafkaUtils.createStream(ssc, zkQuorum, "cyd-demo-azureactivity-streaming-consumer", {topic: 1})

当 Kafka 流实时运行时，我看到 spark 拉数据，但是如果我在 Spark 前一个小时启动 Kafka，它不会提取一小时前的数据。

这是预期的还是有办法在配置中进行设置？

代码运行使用：

sudo $SPARK_HOME/spark-submit --master local[2] --jars /home/steven/jars/elasticsearch-hadoop-6.3.2.jar,/home/steven/jars/spark-streaming-kafka-0-8-assembly_2.11-2.3.1.jar /home/steven/code/demo/test.py
4

1 回答 1

3

如果您总是需要从一开始就拉取数据，那么您需要将 kafka 属性“ auto.offset.reset”设置为“ earliest”。这将从一开始就提取记录。

这个参数是 Kafka 消费者配置 - http://kafka.apache.org/documentation.html#newconsumerconfigs

参考链接 - https://spark.apache.org/docs/2.3.0/streaming-kafka-0-10-integration.html

createStream 有多种实现。您可以使用可以传递 kafka 配置的一种。创建流的示例代码 -

val kafkaParams = Map(
"zookeeper.connect" -> "zookeeper1:2181",
"group.id" -> "spark-streaming-test",
"auto.offset.reset" -> "earliet"
)

val inputTopic = "input-topic"

val stream =  KafkaUtils.createStream(ssc, kafkaParams, Map(inputTopic-> 1), StorageLevel.MEMORY_ONLY_SER)
于 2018-08-20T12:56:08.690 回答