Databricks DLT跨Schema CDC批量处理问题：SCD1表同步报错求助

错误原因分析 #

你遇到的两类错误根源如下：

1. AttributeError: 'function' object has no attribute '_get_object_id'：DLT的装饰器（如@dlt.view）需要在模块顶级作用域注册对象，函数内部定义的装饰器函数无法被DLT的元编程系统正确识别，导致对象ID获取失败。
2. AnalysisException: Failed to read dataset：使用spark.sql静态读取外部表不符合DLT的流处理模式，且DLT管道无法将非管道内定义的静态表识别为合法数据源；同时代码中存在字符串语法错误（未闭合引号）、硬编码视图名重复等问题。

解决方案代码 #

以下是修正后的批量处理代码，通过编程式DLT API实现动态遍历raw层表、应用CDC变更生成SCD1表：

import dlt
from pyspark.sql.functions import *

raw_db_name = "raw_db"
processed_db_name = "processed_db"

# 获取raw_db下所有前缀为raw_的CDC表（可根据实际规则调整过滤条件）
raw_tables = spark.sql(f"SHOW TABLES IN {raw_db_name}") \
    .filter("tableName LIKE 'raw_%'") \
    .select("tableName") \
    .collect()

def process_cdc_table(src_table_name):
    # 生成目标表名（替换raw_前缀为processed_）
    tgt_table_name = src_table_name.replace("raw_", "processed_")
    # 生成唯一视图名，避免跨表冲突
    stream_view_name = f"stream_view_{src_table_name}"
    
    # 1. 编程式创建DLT流视图，读取raw层Delta表的CDC流数据
    dlt.create_view(
        name=stream_view_name,
        comment=f"Streaming CDC data from {raw_db_name}.{src_table_name}",
        spark_conf={"pipelines.incompatibleViewCheck.enabled": "false"}
    )(lambda: spark.readStream.format("delta").table(f"{raw_db_name}.{src_table_name}"))
    
    # 2. 创建processed层目标表（SCD1类型）
    dlt.create_target_table(
        name=f"{processed_db_name}.{tgt_table_name}",
        comment=f"SCD1 cleaned table for {src_table_name}",
        table_properties={
            "quality": "silver",
            "delta.autoOptimize.optimizeWrite": "true"
        }
    )
    
    # 3. 应用CDC变更到目标表，实现SCD1
    dlt.apply_changes(
        target=f"{processed_db_name}.{tgt_table_name}",
        source=stream_view_name,
        keys=["id"],  # 若不同表主键不同，需动态获取主键信息（可通过information_schema查询）
        sequence_by=col("timestamp_ms"),
        apply_as_deletes=expr("op = 'DELETE'"),
        apply_as_truncates=expr("op = 'TRUNCATE'"),
        except_column_list=["op", "timestamp_ms"],  # 排除CDC操作标识和序列列
        stored_as_scd_type=1
    )

# 遍历所有raw层CDC表，批量执行处理
for table_row in raw_tables:
    process_cdc_table(table_row["tableName"])

关键优化点 #

• 编程式DLT API：使用dlt.create_view代替装饰器，允许在循环/函数中动态注册DLT对象，解决作用域识别问题。
• 流读取数据源：用spark.readStream读取raw层Delta表的流数据，符合DLT的增量处理模式，确保管道能识别数据源。
• 动态命名规则：为每个表生成唯一的视图名和目标表名，避免命名冲突。
• 跨Schema目标表：在create_target_table和apply_changes中明确指定processed_db_name，确保表创建在目标Schema下。

扩展说明 #

如果不同表的主键不一致，可通过查询information_schema动态获取主键信息，示例：

def get_table_primary_key(db_name, table_name):
    pk_cols = spark.sql(f"""
        SELECT column_name
        FROM {db_name}.information_schema.key_column_usage
        WHERE table_name = '{table_name}' AND constraint_name LIKE '%primary%'
    """).select("column_name").collect()
    return [row["column_name"] for row in pk_cols]

在process_cdc_table中替换keys=["id"]为keys=get_table_primary_key(raw_db_name, src_table_name)即可。

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