如何将流数据与缓慢更新的表（例如每天一次）连接起来？答案

【问题标题】：How to join stream data with a table updated slowly (e.g. once a day)?如何将流数据与缓慢更新的表（例如每天一次）连接起来？
【发布时间】：2023-04-10 19:05:01
【问题描述】：

在结构化流中，我需要将流数据与一些变化缓慢的数据连接起来。缓慢变化的数据每天都会更新，并且可能不会在固定时间更新。但是，流数据以秒为单位。如果我不想在每个微批次中加载慢变数据，并且希望在更新后获得最新版本的慢变数据，有没有办法做到这一点？

谢谢

【问题讨论】：

标签： apache-spark spark-structured-streaming

【解决方案1】：

我建议在流数据上使用DataStreamWriter.foreachBatch，并在需要时简单地使用cache 和unpersist 变化缓慢的数据集。由于foreachBatch 在驱动程序上工作（虽然在单独的线程上）它应该工作。

在我看来，一种非常先进的方法是开发一个自定义数据源，它可以自己完成“减速部分”。

【讨论】：

谢谢雅克。我希望缓存可以在其源更改后自动更新，有点像增量缓存（似乎开源还不支持它）。现在我正在考虑首先缓存缓慢变化的数据并在每个微批次中读取一些时间（如文件最后修改日期）。如果时间改变，则刷新缓存。
很有意义。请向我们更新您的发现。如果它对您有用，请接受答案。谢谢漂亮！
我可以查询单个文件看是否有变化来决定是否应该重新加载缓存。理想情况下，我应该将当前修改时间与记录时间进行比较。然而，这个记录的时间必须是容错的。执行 mapGroupsWithState 之类的操作可能过于繁重。对于我的场景，我知道该文件每天都会被修改。所以我可以将上次修改日期与当前日期进行比较。如果是当前日期，那么有变化，我应该更新缓存，否则就使用当前缓存。
@processadd 你能显示代码（作为单独的答案）吗？

【解决方案2】：

我试过这个。我将单个文件的最后修改时间存储在一个变量中，然后广播它，然后在 foreachBatch 中我再次得到这个时间。如果它们不同，我可以刷新缓存。然后我发现我不必广播变量。如果变量在 foreachBatch 之前获取了值，它仍然会在 foreachBatch 中保留原始值（对于在 intelliJ 中运行的本地模式）。代码如下：

var latestModified = Files.getLastModifiedTime(Paths.get("/some_file"))
var deltaTable = DeltaTable.forPath(spark, deltaPath)
var c = deltaTable.toDF.cache()
df
.writeStream
......
.foreachBatch { (df, batchId) =>
    val currentModifiedTime = Files.getLastModifiedTime(Paths.get("/some_file"))
    if (!currentModifiedTime.equals((latestModified))){
          c.unpersist()
          c = deltaTable.toDF.cache()
          latestModified = currentModifiedTime
        }
... ...

【讨论】：