Spark Streaming Kafka - 如何在处理所有现有消息后停止流式传输（优雅地）答案

【问题标题】：Spark Streaming Kafka - How to stop streaming after processing all existing messages (gracefully)Spark Streaming Kafka - 如何在处理所有现有消息后停止流式传输（优雅地）
【发布时间】：2020-08-15 09:56:10
【问题描述】：

这就是我想要做的事情

来自 kafka 主题的流数据，该主题不断获取数据。每天运行两次作业，以处理此时所有数据现有数据并停止流。

所以我最初在查询上放置并调用停止，但它抛出“TimeoutException”

然后我尝试动态增加超时，但现在我得到 java.io.IOException: Caused by: java.lang.InterruptedException

那么，有没有什么方法可以优雅地停止流而不会出现任何异常？

下面是我当前的代码（部分），它抛出了中断的异常

df = (
    spark.readStream.format("kafka")
    .option("kafka.bootstrap.servers", os.environ["KAFKA_SERVERS"])
    .option("subscribe", config.kafka.topic)
    .option("startingOffsets", "earliest")
    .option("maxOffsetsPerTrigger", 25000)
    .load()
)


#   <do some processing and save the data>
def save_batch(batch_df, batch_id):
    pass

query = df.writeStream.foreachBatch(save_batch).start(
    outputMode="append",
    checkpointLocation=os.path.join(checkpoint_path, config.kafka.topic),
)

while query.isActive:
    progress = query.lastProgress
    if progress and progress["numInputRows"] < 25000 * 0.9:
        timeout = sum(progress["durationMs"].values())
        timeout = min(5 * 60 * 1000, max(15000, timeout))
        spark.conf.set("spark.sql.streaming.stopTimeout", str(timeout))
        stream_query.stop()
        break
    time.sleep(10)

Spark 版本：2.4.5 斯卡拉版本：2.1.1

【问题讨论】：

即使将指定的配置设置为 true，它仍然会抛出 InterruptedException
您是否考虑过使用更简单的 Spark 批处理作业而不是流式作业并每天安排两次？这个用例听起来不适合流式传输给我。
kafka 主题填充了大量数据，并且 spark 流处理了检查点并解决了头痛问题，我认为如果尝试做一些批处理作业，我必须管理它，而这并没有让它变得简单。如果我的假设是错误的，请告诉我
我同意@mike。如果你真的想这样做，我会使用Accumulator 来跟踪最后一个活动批次的时间戳，然后在isActive 中取回时间戳，如果时间戳>限制，则正常关闭
再一次如何优雅地关闭，query.stop() 正在抛出 TimeoutException 或 InterruptedException

标签： apache-spark pyspark apache-kafka

【解决方案1】：

更新：在 Spark 3.3 中，.trigger(availableNow=True) 是一个可以很好地与 .option("maxOffsetsPerTrigger", 25000) 配合使用的选项。

我会推荐.trigger(once=True) 和.awaitTermination() (https://spark.apache.org/docs/latest/structured-streaming-programming-guide.html#triggers)。

警告：这不适用于.option("maxOffsetsPerTrigger", 25000)，但如果未设置maxOffsetsPerTrigger，它将默认拉取自上次运行以来的所有偏移量以创建一个大型微批次。

df = spark \
    .readStream \
    .format("kafka") \
    .option("kafka.bootstrap.servers", os.environ["KAFKA_SERVERS"]) \
    .option("subscribe", config.kafka.topic) \
    .option("startingOffsets", "earliest") \
    .load()

def foreach_batch_function(df, epoch_id):
    # Transform and write batchDF
    pass

df \
    .writeStream \
    .foreachBatch(foreach_batch_function) \
    .trigger(once=True) \
    .start(
        outputMode="append",
        checkpointLocation=os.path.join(checkpoint_path, config.kafka.topic),
    ) \
    .awaitTermination()

【讨论】：