自动加载器过滤重复项

Question

我正在处理流数据帧，想知道如何消除重复项并仅选择最新的

modifiedon

行。

例如。

id	修改于
1	2023/03/08
1	2023/03/08
2	2023年2月8日
2	2023/03/08

想要的df

id	修改于
1	2023/03/08
2	2023/03/08

因此，如果存在完全相同（100％）相同的行，例如 id 1，则需要删除所有行并仅为每个 ID 保留 1 行。而且，如果同一 ID 出现多次（更新多次），我需要根据最大修改列进行选择。

df = (
    spark.readStream
    .option("delimiter", ",")
    .option("quote", '"')
    .option("mode", "permissive")
    .option("lineSep", "\r\n")
    .option("multiLine", "true")
    .format("cloudFiles")
    .option("cloudFiles.format", source_format)
    .option("cloudFiles.schemaLocation", checkpoint_directory)
    .option("header", "false")
    .option("escape", '"')
    .schema(schema)
    .load(data_source)
)

Answer 1

您可以按照以下方法进行操作。

以下是我使用的数据。

df = df.dropDuplicates().withColumn("modifiedon",to_timestamp("modifiedon","dd/MM/yyyy"))
df = df.withWatermark("modifiedon", "1 day")
df = df.groupBy("id").agg(max("modifiedon").alias("modifiedon"))
display(df)

此处，在 1 天的时间内对流数据应用聚合并在列上执行最大值

modifiedon

。

输出：