Databricks中通过DLT流水线动态执行SCD2表加载

1. 配置DLT流水线的「Configuration」项 #

将你的SCD2表配置JSON作为单个键值对添加到流水线配置中：

• 键：自定义一个标识（比如scd2.table.configs）
• 值：你的完整JSON字符串（直接在UI输入框中粘贴即可，格式保持正确）

示例配置项：

scd2.table.configs={"object1": {"object_name": "table", "business_key_column": "table_bk", "except_column_list": "change_date", "sequence_column": "change_date"}, "object2": {...}}

2. 在DLT笔记本中读取并解析配置 #

通过spark.conf获取配置值，解析为可操作的字典后，就能遍历配置动态加载SCD2表：

import json
import dlt

# 获取流水线配置中的SCD2表配置字符串
scd2_config_str = spark.conf.get("scd2.table.configs")
# 解析为字典格式
scd2_config = json.loads(scd2_config_str)

# 遍历每个表配置，执行SCD2加载逻辑
for _, table_settings in scd2_config.items():
    table_name = table_settings["object_name"]
    business_key = table_settings["business_key_column"]
    exclude_cols = table_settings["except_column_list"].split(",")
    sequence_col = table_settings["sequence_column"]

    # 创建SCD2目标表
    dlt.create_streaming_table(f"{table_name}_scd2")
    # 应用增量变更实现SCD2
    dlt.apply_changes(
        target=f"{table_name}_scd2",
        source=f"your_source_stream_{table_name}",  # 替换为实际的数据源流
        keys=[business_key],
        sequence_by=sequence_col,
        except_columns=exclude_cols,
        stored_as_scd_type=2
    )

注意事项 #

• 如果配置项可能不存在，可添加默认值避免报错：spark.conf.get("scd2.table.configs", "{}")
• 若需单独读取某个配置项（而非整体JSON），可在流水线Configuration中拆分键值对（比如object1.object_name=table），直接用spark.conf.get("object1.object_name")获取

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