如何使用 Trigger.Once 选项在 Spark 3 Structured Streaming Kafka/文件源中配置背压答案

【问题标题】：How to configure backpreasure in Spark 3 Structure Stream Kafka/Files source with Trigger.Once option如何使用 Trigger.Once 选项在 Spark 3 Structured Streaming Kafka/文件源中配置背压
【发布时间】：2023-03-20 19:57:01
【问题描述】：

In Spark 3 Behave of backpressure option on Kafka and File Source for trigger.once scenario was changed.

但我有一个问题。当我想使用 TriggerOnce 时，如何为我的工作配置背压？

在 spark 2.4 中，我有一个用例，用于回填一些数据，然后启动流。所以我使用了一次触发器，但我的回填场景可能非常非常大，有时会因为洗牌和驱动程序内存而在我的磁盘上产生太大的负载，因为 FileIndex 缓存在那里。所以我使用 max maxOffsetsPerTrigger 和 maxFilesPerTrigger 来控制我的 spark 可以处理多少数据。这就是我配置背压的方式。

现在你移除了这个能力，所以假设有人可以提出一个新的方法吗？

【问题讨论】：

标签： apache-spark spark-structured-streaming spark-kafka-integration

【解决方案1】：

Trigger.Once 现在会忽略这些选项（在 Spark 3 中），因此它总是会在第一次加载时读取所有内容。

您可以解决该问题 - 例如，您可以将触发器设置为周期性启动流，使用某个值（例如 1 小时），并且不执行 .awaitTermination，但有一个并行循环将检查第一批是否完成，并停止流。或者您可以将其设置为连续模式，然后检查批次是否为 0 行，然后终止流。初始加载后，您可以将流切换回 Trigger.Once

【讨论】：

这是一个非常肮脏的解决方案，给我的代码带来了一些多线程复杂性。我还可以在每第二批中抛出异常，并在此特定异常上处理流关闭。但我仍然希望在没有 HACK 的情况下学习如何做到这一点。
我不敢相信 Spark 开发者一次性删除了 Trigger 的背压功能，并且没有提供任何好的替代品。
你确定它以前有效吗？在这个解决方案中你不需要做任何多线程，只是不要使用awaitTermination
我知道开发团队正在为此努力......但不知道它将登陆哪个 Spark 版本
Previous Trigger.Once 只是生成一个批次。此批次以与任何其他触发器相同的方式尊重所有配置。现在我需要做一个肮脏的黑客，当第一批通过时停止。主要问题是，如果我在批处理已经开始时失败，则会出现偏移 WAL，这对我来说是个问题。因为我无法覆盖下一批的配置。