azure - 自动加载器过滤重复项

Question

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

例如。

<表类=“s-表”><标题>id修改 <正文>12023年3月8日12023年3月8日22023年2月8日22023年3月8日

所需的 df

<表类=“s-表”><标题>id修改 <正文>12023年3月8日22023年3月8日

因此，如果存在完全相同(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 执行 max 操作。

输出: