防止 Spark 在流/流连接中存储状态

Question

我有两个流数据集，我们将它们称为fastStream和slowStream。

这fastStream是我通过结构化流 API 从 Kafka 消费的流数据集。我预计每秒可能会收到数千条消息。

这slowStream实际上是一个引用（或查找）表，正在由另一个流“更新插入”，并且包含我想要在将fastStream记录保存到表之前加入到每条消息的数据。仅当有人更改元数据时才会slowStream更新，这可能随时发生，但我们预计可能每隔几天更改一次。

中的每条记录都fastStream将具有一条对应的消息slowStream，我本质上希望使该连接立即与表中的任何数据发生slowStream。我不想等到新数据到达时看看是否会发生潜在的匹配slowStream。

我遇到的问题是，根据 Spark 文档：

因此，对于这两个输入流，我们将过去的输入缓冲为流状态，以便我们可以将每个未来的输入与过去的输入进行匹配，并相应地生成连接结果。

我尝试添加水印，fastStream但我认为它没有效果，因为文档表明需要在连接中引用带水印的列

理想情况下我会写这样的东西：

# Apply a watermark to the fast stream
fastStream = spark.readStream \
.format("delta") \
.load("dbfs:/mnt/some_file/fastStream") \
.withWatermark("timestamp", "1 hour") \
.alias("fastStream")

# The slowStream cannot be watermarked since it is only slowly changing
slowStream = spark.readStream \
.format("delta") \
.load("dbfs:/mnt/some_file/slowStream") \
.alias("slowStream")

# Prevent the join from buffering the fast stream by 'telling' spark that there will never be new matches.
fastStream.join( 
  slowStrean,
  expr(""" 
    fastStream.slow_id = slowStream.id
    AND fastStream.timestamp > watermark
    """
  ),
  "inner"
).select("fastStream.*", "slowStream.metadata")

但我不认为你可以watermark在 SQL 表达式中引用 the 。

本质上，虽然我很高兴有slowStream缓冲（因此整个表都在内存中），但我无法缓冲fastStream，因为该表将很快消耗所有内存。相反，我只想删除fastStream不匹配的消息，而不是保留它们以查看它们将来是否匹配。

非常感谢任何帮助。

Answer 1

对于内部流-流连接，水印和事件时间约束（连接条件）是可选的。

如果无界状态在体积方面对您来说不是问题，您可以选择不指定它们。在这种情况下，所有数据都将被缓冲，来自 fastStream 的数据将立即与来自 SlowStream 的所有数据合并。

只有当两个参数都指定时，您的状态才会被清除。请注意这两个参数的用途：

• 事件时间约束（时间范围连接条件）：两个事件在各自源处生成之间的最大时间范围是多少？

• 水印：事件在源和处理引擎之间传输时可以延迟的最大持续时间是多少？

要定义这两个参数，您需要首先回答上述问题（引自 O`Reilly 出版的《学习 Apache Spark，第 2 版》一书）。

关于您的代码评论：

“通过‘告诉’spark永远不会有新的匹配来防止连接缓冲快速流。”

请记住，流-流连接中的缓冲是必要的。否则，您将只能加入当前微批次中可用的数据。由于 SlowStream 没有定期更新，但 fastStream 更新数据的速度非常快，如果不缓冲数据，您可能永远不会获得任何连接匹配。

总的来说，对于您所描述的用例（“将快速变化的数据与缓慢变化的元数据连接起来”），通常最好使用流静态连接方法，其中缓慢变化的数据成为静态部分。

在流静态连接中，流数据中的每一行都将与完整的静态数据连接，而静态表则在每个微批次中加载。如果加载静态表会降低性能，您可能会考虑缓存它并定期更新它，如Stream-Static Join：如何定期刷新（取消持久/持久）静态数据帧中所述。