如何在结构化流中终止dropDuplicates状态以避免OOM？

Question

我希望使用星火结构化流计算每天的唯一访问，因此我使用以下代码

.dropDuplicates("uuid")

在第二天，维护今天的状态应该取消，这样我就可以得到第二天唯一访问的正确计数，并避免OOM。星星之火文档使用带有水印的dropDuplicates来指示，例如：

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

但是水印列必须在dropDuplicates中指定。在这种情况下，uuid和时间戳将被用作使用相同的uuid和时间戳去重复元素的组合键，这与我所期望的不一样。

那么有一个完美的解决方案吗？

Answer 1

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

在研究水印和dropDuplicates的源代码时，我发现除了eventTime列之外，水印还支持窗口列，因此我们可以使用以下代码：

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

因为同一天中的所有事件都有相同的窗口，这将产生与仅使用uuid去重复相同的结果。希望能帮助一个人。

Answer 2

下面是火花文件中提议的程序的修改。技巧是操纵事件时间，即将事件时间放入桶中。假设事件时间以毫秒为单位。

// removes all duplicates that are in 15 minutes tumbling window.
// doesn't remove duplicates that are in different 15 minutes windows !!!!
public static Dataset<Row> removeDuplicates(Dataset<Row> df) {
    // converts time in 15 minute buckets
    // timestamp - (timestamp % (15 * 60))
    Column bucketCol = functions.to_timestamp(
            col("event_time").divide(1000).minus((col("event_time").divide(1000)).mod(15*60)));
    df = df.withColumn("bucket", bucketCol);

    String windowDuration = "15 minutes";
    df = df.withWatermark("bucket", windowDuration)
            .dropDuplicates("uuid", "bucket");

    return df.drop("bucket");
}

Answer 3

我发现窗口函数不起作用，所以我选择使用window.start或window.end。

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