基于Databricks Autoloader与ForEachBatch的变更追踪异常排查

我使用Databricks Autoloader的Trigger Once模式从S3存储位置加载Parquet文件，目标是通过对比源表与目标Delta表实现变更数据捕获（CDC），识别并记录INSERT、UPDATE、DELETE操作，而非执行MERGE操作，旨在构建一张仅追加的变更日志表。当前的ForEachBatch函数可正常运行，但无法捕获任何UPDATE或DELETE操作。

输入数据示例 #

Batch #1 #

batch_1_timestamp = "2024-07-19"
data = [
("1", "John", batch_1_timestamp),
("2", "David", batch_1_timestamp),
]

Batch #2 #

batch_2_timestamp = "2024-07-20"
data = [
("1", "John", batch_2_timestamp), # 无变更记录
("2", "David Jones", batch_2_timestamp), # 更新记录
("3", "Bob", batch_2_timestamp),   # 新增记录
]

Batch #3 #

batch_3_timestamp = "2024-07-21"
data = [
("2", "David", batch_3_timestamp), # 更新记录
("3", "Bob", batch_3_timestamp), # 无变更记录
# 记录1已删除
]

预期输出 #

当前使用的函数代码 #

def upsertToDelta(batch_df: DataFrame, batch_id: int):
    target_table = (
        spark
        .table(target_table_qualified)
        )

    hashed_df = batch_df
    df_source = utl.add_row_hash_to_dataframe(
        df=hashed_df,
        ignore_columns=metadata_columns
    )
    
    created_at_timestamp = datetime.now()

    new_records = df_source.alias("s").join(
        target_table.alias("t"),
        on=[
            col("s.id") == col("t.id")
        ],
        how = "left"
    ).filter(
        col("t.id").isNull()
    ).select(
        "s.*",
        lit("INSERT").alias("operation_type"),
        lit(created_at_timestamp).alias("created_at"),
    )
    
    updated_records = df_source.alias("s").join(
        target_table.alias("t"),
        on=[
            col("s.id") == col("t.id"),
            col("s.batch_timestamp") > col("t.batch_timestamp")
        ],
        how="inner"
    ).filter(
        col("s.row_hash") != col("t.row_hash")
    ).select(
        "s.*",
        lit("UPDATE").alias("operation_type"),
        lit(created_at_timestamp).alias("created_at")
    )
    
    deleted_records = (
        target_table.alias("t")
        .join(
        df_source.alias("s"),
        on=[
            col("t.id") == col("s.id"),
        ],
        how="left_anti"
    )
        .select('t.*')
        .withColumn("operation_type", lit("DELETE"))
        .withColumn("created_at", lit(created_at_timestamp))
    )

    df_processed = new_records.union(updated_records).union(deleted_records)

    df_processed.write.saveAsTable(
        target_table_qualified,
        format="delta",
        mode="append",
    )
    

# Write file stream
write_stream = (df_converted.writeStream
                .format("delta")
                .outputMode("append")
                .option("cloudFiles.schemaEvolutionMode", "rescue")
                .foreachBatch(upsertToDelta)
                .queryName(f"BronzeMerge[{target_table_qualified}]")
                .trigger(once=True)
                .option("checkpointLocation", checkpoint_location)
                .start())
write_stream.awaitTermination()

问题原因与修复方案 #

核心问题 #

1. UPDATE逻辑错误：当前代码直接用变更日志表作为target_table进行对比，但日志表是仅追加的全量历史记录，而非存储每个ID最新状态的业务表，导致无法准确识别真实的更新操作。
2. DELETE逻辑错误：同样用日志表做对比，日志表中包含历史删除记录，无法正确找出当前批次中缺失的业务数据。

修复思路 #

要实现正确的CDC日志，需要拆分两张表：

• 业务状态表：存储每个id的最新业务状态，用于和新批次数据对比识别变更
• 变更日志表：仅追加记录所有INSERT/UPDATE/DELETE操作（即原目标表）

修复后的代码 #

# 定义业务状态表和变更日志表的全限定名称
state_table_qualified = "your_database.your_state_table"
target_table_qualified = "your_database.your_cdc_log_table"

def upsertToDelta(batch_df: DataFrame, batch_id: int):
    # 获取当前业务状态表的最新数据
    state_table = spark.table(state_table_qualified)
    
    # 给源数据添加行哈希（忽略元数据列）
    df_source = utl.add_row_hash_to_dataframe(
        df=batch_df,
        ignore_columns=metadata_columns
    )
    
    created_at_timestamp = datetime.now()
    
    # 识别新增记录：源表存在、状态表不存在的ID
    new_records = df_source.alias("s")\
        .join(state_table.alias("t"), on="id", how="left")\
        .filter(col("t.id").isNull())\
        .select(
            "s.*",
            lit("INSERT").alias("operation_type"),
            lit(created_at_timestamp).alias("created_at")
        )
    
    # 识别更新记录：ID匹配但业务数据哈希不同，且源批次时间更新
    updated_records = df_source.alias("s")\
        .join(state_table.alias("t"), on="id", how="inner")\
        .filter(
            (col("s.row_hash") != col("t.row_hash")) &
            (col("s.batch_timestamp") > col("t.batch_timestamp"))
        )\
        .select(
            "s.*",
            lit("UPDATE").alias("operation_type"),
            lit(created_at_timestamp).alias("created_at")
        )
    
    # 识别删除记录：状态表存在、源批次不存在的ID，用当前处理日期作为批次时间戳
    deleted_records = state_table.alias("t")\
        .join(df_source.alias("s"), on="id", how="left_anti")\
        .select(
            "t.id",
            "t.name",
            lit(created_at_timestamp.strftime("%Y-%m-%d")).alias("batch_timestamp"),
            lit("DELETE").alias("operation_type"),
            lit(created_at_timestamp).alias("created_at")
        )
    
    # 合并所有变更记录写入日志表
    df_processed = new_records.union(updated_records).union(deleted_records)
    df_processed.write.mode("append").saveAsTable(target_table_qualified)
    
    # 更新业务状态表：MERGE保留每个ID的最新状态
    df_source.createOrReplaceTempView("temp_source")
    spark.sql(f"""
        MERGE INTO {state_table_qualified} t
        USING temp_source s
        ON t.id = s.id
        WHEN MATCHED THEN UPDATE SET *
        WHEN NOT MATCHED THEN INSERT *
    """)

# 初始化业务状态表（首次运行时执行）
if not spark.catalog.tableExists(state_table_qualified):
    initial_schema = batch_df.schema
    spark.createDataFrame([], schema=initial_schema).write.saveAsTable(state_table_qualified)

# 启动流处理
write_stream = (df_converted.writeStream
                .format("delta")
                .outputMode("append")
                .option("cloudFiles.schemaEvolutionMode", "rescue")
                .foreachBatch(upsertToDelta)
                .queryName(f"CDC_Log_Generator[{target_table_qualified}]")
                .trigger(once=True)
                .option("checkpointLocation", checkpoint_location)
                .start())
write_stream.awaitTermination()

关键说明 #

• 业务状态表维护：必须单独维护每个ID的最新状态，这是准确识别变更的核心前提。
• 删除时间戳处理：由于源批次不包含删除记录，用当前处理日期作为batch_timestamp，匹配预期输出格式。
• 行哈希校验：确保utl.add_row_hash_to_dataframe仅计算name等业务字段的哈希，忽略batch_timestamp这类元数据列，避免误判更新。

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