Databricks DLT处理Kafka CDC数据时如何实现表数据追加而非覆盖？

问题根源 #

你现在的DLT代码用@dlt.table()直接返回当日处理后的DataFrame，默认会执行CREATE OR REPLACE TABLE，所以每天运行都会覆盖旧数据。要实现追加或CDC合并更新，得调整DLT表的写入逻辑，结合CDC的操作类型（op字段）做增量合并，而非全量替换。

核心改进方向 #

• 替换全量逻辑：用DLT的merge功能，针对CDC里的op字段（'d'=删除，其他=插入/更新），对RAW层现有表做增量合并，兼顾新增、更新、删除操作。
• 适配批处理：你是每日触发批处理，完全不需要read_stream/spark.readStream，直接处理当日Landing层的批次数据即可。
• 优化Schema推导：去掉collect()拿schema的逻辑，避免当日无数据时报错，改用更可靠的推导方式。

修改后的代码示例 #

import json
import dlt
from pyspark.sql.functions import from_json, col
from pyspark.sql.types import StructType

def generate_cdc_table(table_name, source_df):
    # 可靠推导CDC的value字段Schema，避免空数据报错
    cdc_schema = spark.read.json(source_df.rdd.map(lambda row: row.value)).schema
    
    # 解析CDC数据，带上Kafka元数据方便排查
    parsed_df = source_df.withColumn("value", from_json("value", cdc_schema)) \
        .select(
            col("value.*"),
            col("offset").alias("_kafka_offset"),
            col("timestamp").alias("_kafka_timestamp")
        )
    
    # 区分删除/插入更新数据
    delete_df = parsed_df.filter(col("op") == "d").select(col("before.*"), col("op"))
    upsert_df = parsed_df.filter(col("op") != "d").select(col("after.*"), col("op"))
    processed_df = upsert_df.unionByName(delete_df, allowMissingColumns=True)
    
    # 定义DLT表，用merge实现增量更新
    @dlt.table(
        name=table_name,
        comment=f"CDC raw data for {table_name}",
        path=f"{raw_file_path}/{table_name}",
        table_properties={
            "delta.enableChangeDataFeed": "true",  # 可选：开启CDF方便后续层使用
            "delta.appendOnly": "false"  # 允许更新/删除操作
        }
    )
    @dlt.merge(
        target=f"live.{table_name}",
        condition="target.id = source.id"  # 替换为你的表主键，比如user_id/order_id
    )
    def create_or_merge_table():
        return processed_df
    
    # 设置Merge规则
    create_or_merge_table.when_matched_delete(condition="source.op = 'd'") \
        .when_matched_update_all(condition="source.op != 'd'") \
        .when_not_matched_insert_all(condition="source.op != 'd'")

# 遍历处理每张表
for tbl_name in table_column_schema:
    # 读取当日Landing层的Delta表数据
    landing_df = spark.sql(f"SELECT value, offset, timestamp FROM {db}.{tbl_name}")
    
    # 生成增量合并的RAW层表
    generate_cdc_table(tbl_name, landing_df)

关键说明 #

• Merge逻辑细节：
  • 主键匹配且op='d'：删除目标表对应行
  • 主键匹配且op!='d'：更新目标表对应行
  • 主键不匹配且op!='d'：插入新数据
• 主键必须替换：把代码里的target.id = source.id换成每张表实际的主键字段，否则Merge逻辑无法正常工作
• 空数据防护：可以添加判断if landing_df.count() == 0则跳过该表，避免无数据时无效运行
• 性能优化：可以给RAW层表按_kafka_timestamp或日期分区，提升后续查询效率

额外建议 #

1. 添加日志记录每张表处理的行数，方便问题排查
2. 增加Schema校验逻辑，确保CDC数据字段与目标表匹配，避免Schema漂移
3. 保留Kafka的offset和时间戳字段，便于后续回溯源数据

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