apache-spark - 如何使结构化流中的 dropDuplicates 状态过期以避免 OOM？

Question

我想使用 spark 结构化流计算每天的唯一访问次数，因此我使用以下代码

.dropDuplicates("uuid")

并且在第二天应该删除今天维护的状态，以便我可以获得第二天的唯一访问的正确计数并避免 OOM。 spark 文档指示使用带有水印的 dropDuplicates，例如:

.withWatermark("timestamp", "1 day")
.dropDuplicates("uuid", "timestamp")

但水印列必须在 dropDuplicates 中指定。在这种情况下，uuid 和时间戳将用作组合键，以对具有相同 uuid 和时间戳的元素进行重复数据删除，这不是我所期望的。

那么有没有完美的解决方案呢？

Answer 1

经过几天的努力，我终于找到了自己的方法。

在研究watermark和dropDuplicates的源码时，发现watermark除了eventTime列之外，还支持window列，所以我们可以使用如下代码:

.select(
    window($"timestamp", "1 day"),
    $"timestamp",
    $"uuid"
  )
.withWatermark("window", "1 day")
.dropDuplicates("uuid", "window")

由于同一天的所有事件都具有相同的窗口，因此这将产生与仅使用 uuid 进行重复数据删除相同的结果。希望可以帮助某人。