如何在Databricks Autoloader中实现去重并保留最新记录？

问题分析 #

你需要在流处理场景下（Autoloader）为每个ID保留最新timestamp的记录，批处理中常用的rank窗口函数无法跨微批维护状态，而withWatermark + dropDuplicates只能过滤同ID同时间的重复，无法处理跨多天的更新数据。以下是两种可行的解决方案：

方案一：使用mapGroupsWithState维护自定义状态 #

这种方式通过流处理的状态管理API，为每个ID持久化存储最新记录，支持跨任意时间间隔的更新。

步骤1：导入依赖并定义Schema #

from pyspark.sql import functions as F
from pyspark.sql.types import StructType, StructField, StringType, TimestampType, IntegerType
from pyspark.sql.streaming import GroupState, GroupStateTimeout
from pyspark.sql import Row

# 替换为你的实际数据Schema
input_schema = StructType([
    StructField("ID", StringType(), nullable=False),
    StructField("timestamp", TimestampType(), nullable=False),
    StructField("value", IntegerType(), nullable=True)
])

# 状态Schema与输入数据一致，用于存储每个ID的最新记录
state_schema = input_schema

步骤2：定义状态更新函数 #

该函数会对比当前微批数据与已存储的状态，保留每个ID的最新记录：

def update_latest_record(key: str, values: Iterator[Row], state: GroupState[Row]) -> Iterator[Row]:
    # 读取当前状态中存储的最新记录
    current_latest = state.getOption().getOrElse(None)
    
    # 遍历当前微批的所有记录，更新为最新版本
    for value in values:
        if current_latest is None or value.timestamp > current_latest.timestamp:
            current_latest = value
    
    # 更新状态并输出最新记录
    if current_latest is not None:
        state.update(current_latest)
        # 可选：设置空闲超时（如30天），清理长时间无更新的ID状态，防止内存溢出
        state.setTimeoutDuration("30 days")
        yield current_latest

步骤3：结合Autoloader读取并处理流数据 #

# 用Autoloader读取文件流（替换为你的文件格式和路径）
stream_df = spark.readStream \
    .format("cloudFiles") \
    .option("cloudFiles.format", "csv")  # 支持parquet、json等格式
    .option("cloudFiles.schemaLocation", "/dbfs/path/to/schema_store")  # 自动推断Schema的存储路径
    .schema(input_schema) \
    .load("/dbfs/path/to/source_files")

# 按ID分组，应用状态更新逻辑
latest_records_df = stream_df \
    .groupByKey(lambda row: row.ID) \
    .mapGroupsWithState(
        outputMode="update",  # 仅输出状态有变化的记录（性能更优）
        timeoutConf=GroupStateTimeout.IdleTimeout,  # 启用空闲超时
        stateSchema=state_schema
    )(update_latest_record)

# 输出结果（示例：写入Delta表）
query = latest_records_df.writeStream \
    .format("delta") \
    .option("checkpointLocation", "/dbfs/path/to/checkpoint") \
    .table("your_target_delta_table")

方案二：使用Delta Lake的MERGE操作 #

如果需要将最新记录持久化到Delta表，且需要ACID保障，可通过foreachBatch执行MERGE逻辑，自动覆盖旧版本数据：

实现代码 #

# 读取Autoloader流数据
stream_df = spark.readStream \
    .format("cloudFiles") \
    .option("cloudFiles.format", "parquet") \
    .option("cloudFiles.schemaLocation", "/dbfs/path/to/schema_store") \
    .load("/dbfs/path/to/source_files")

# 定义MERGE逻辑：当ID匹配且新记录时间戳更新时覆盖，否则插入
def merge_updates(micro_batch_df, batch_id):
    micro_batch_df.createOrReplaceTempView("stream_updates")
    
    spark.sql("""
        MERGE INTO your_target_table t
        USING stream_updates u
        ON t.ID = u.ID
        WHEN MATCHED AND u.timestamp > t.timestamp THEN
          UPDATE SET *
        WHEN NOT MATCHED THEN
          INSERT *
    """)

# 启动流处理，执行MERGE
query = stream_df.writeStream \
    .foreachBatch(merge_updates) \
    .option("checkpointLocation", "/dbfs/path/to/checkpoint") \
    .start()

两种方案对比 #

内容的提问来源于stack exchange，提问作者WilliamEllisWebb