scala - 即使设置了 auto.offset.reset=latest，Spark 结构化流查询也始终以 auto.offset.rest=earliest 开头

Question

我在尝试使用 Spark 结构化流从 Kafka 读取数据时遇到了一个奇怪的问题。
我的用例是能够从可用的最大/最新偏移量中读取主题。

我的读取配置:

val data = spark
     .readStream
     .format("kafka")
     .option("kafka.bootstrap.servers", "some xyz server")
     .option("subscribe", "sampletopic")
     .option("auto.offset.reset", "latest")
     .option("startingOffsets", "latest")
     .option("kafkaConsumer.pollTimeoutMs", 20000)
     .option("failOnDataLoss","false")
     .option("maxOffsetsPerTrigger",20000)
     .load()

我的写配置:

data
    .writeStream
    .outputMode("append")
    .queryName("test") 
    .format("parquet")
    .option("checkpointLocation", "s3://somecheckpointdir")
    .start("s3://outpath").awaitTermination()

使用的版本:

spark-core_2.11 : 2.2.1

spark-sql_2.11 : 2.2.1

spark-sql-kafka-0-10_2.11 : 2.2.1

我已经在网上和 [Kafka 文档]( https://kafka.apache.org/0100/documentation.html0/

我正在使用新的消费者 apis，并且如文档所示，我只需要将 auto.offset.reset 设置为“最新”或将开始偏移设置为“最新”，以确保我的 Spark 作业从 Kafka 中每个分区可用的最新偏移量开始使用.

我也知道设置 auto.offset.reset仅在第一次启动新查询时启动，而不是在重新启动应用程序时启动，在这种情况下，它将继续从上次保存的偏移量中读取。

我正在使用 s3 检查我的偏移量。我看到它们是在 s3://somecheckpointdir 下生成的。

我面临的问题是 Spark 作业总是从最早的偏移量读取，即使在第一次启动应用程序时在代码中指定了最新选项，我在 Spark 日志中看到了这一点。 auto.offset.reset = earliest正在使用。我还没有看到与此特定问题相关的帖子。

我想知道我是否在这里遗漏了什么，以及之前是否有人见过这种行为。任何帮助/方向确实会很有用。谢谢你。

Answer 1

所有 Kafka 配置都应设置为 kafka.字首。因此正确的选项键是 kafka.auto.offset.reset .

你永远不应该设置 auto.offset.reset .相反，“设置源选项 startingOffsets 来指定从哪里开始。结构化流管理内部消耗哪些偏移量，而不是依赖 kafka 消费者来做。这将确保在新主题/分区时不会丢失任何数据是动态订阅的。请注意，startingOffsets 仅在启动新的流式查询时适用，并且恢复将始终从查询停止的位置开始。” [1]

[1] http://spark.apache.org/docs/latest/structured-streaming-kafka-integration.html#kafka-specific-configurations