Azure Databricks中如何将Silver表新增记录实时转发至SQL Server？

我们正在使用Azure Databricks构建典型的Medallion架构Delta Lake。业务要求记录一到达Delta表就立即转发至SQL Server实例。我们了解过Change Data Feed（变更数据馈送）概念，但它似乎是收集变更数据而非实时转发记录。我们采用Delta Live Tables（DLT）框架实现Bronze表到Silver表的转换，请问是否存在方法可让Silver表的每条记录一到达就立即转发至外部SQL Server实例？

方法1：在DLT Silver表定义中嵌入实时转发逻辑 #

直接在生成Silver表的流式处理过程中，添加foreachBatch逻辑，将每个微批的数据同步写入SQL Server。这种方法能保证Silver表写入与SQL Server转发的原子性（依赖DLT的事务机制），且延迟最低。

示例代码（Python）：

import dlt
from pyspark.sql.functions import current_timestamp, col

# 假设已定义Bronze表
@dlt.table(name="bronze_table", comment="原始数据源")
def bronze_table():
    return spark.readStream.format("cloudFiles")\
        .option("cloudFiles.format", "json")\
        .load("/path/to/raw/data")

# 定义Silver表并同步转发至SQL Server
@dlt.table(name="silver_table", comment="清洗后的结构化数据")
def silver_table():
    # 从Bronze表流式读取并转换得到Silver表数据
    silver_df = dlt.read_stream("bronze_table").select(
        col("id"),
        col("business_value"),
        current_timestamp().alias("processed_at")
    )
    
    # 定义批量写入SQL Server的逻辑
    def batch_write_sql_server(batch_df, batch_id):
        batch_df.write.format("jdbc")\
            .option("url", "jdbc:sqlserver://<你的SQL Server地址>.database.windows.net:1433;databaseName=<数据库名>")\
            .option("dbtable", "<目标表名>")\
            .option("user", "<用户名>")\
            .option("password", "<密码>")\
            .mode("append")\
            .save()
    
    # 启动流式写入SQL Server的任务
    silver_df.writeStream.foreachBatch(batch_write_sql_server).start()
    
    return silver_df

关键注意点：

• 确保DLT管道运行在连续模式（Continuous Mode），可将延迟降低至数百毫秒，接近实时。
• 若需保证事务一致性，可在batch_write_sql_server中添加失败重试逻辑，或利用DLT的自动重试机制（DLT会自动重试失败的微批）。

方法2：利用Delta CDF实时捕获Silver表变更并转发 #

CDF结合流式查询可实现实时转发，适合需要独立控制转发逻辑的场景（比如不需要和Silver表的生成强绑定）。

示例代码（Python）：

# 先确保Silver表已开启CDF，可在DLT表定义中添加table_properties={"delta.enableChangeDataFeed": "true"}

# 流式读取Silver表的CDF变更数据
silver_cdf_stream = spark.readStream.format("delta")\
    .option("readChangeFeed", "true")\
    .option("startingVersion", "latest")\
    .table("silver_table")

# 定义批量写入SQL Server的逻辑，过滤出新增记录（可根据业务需求调整）
def write_cdf_to_sql(batch_df, batch_id):
    # 仅处理新增的记录（若需处理更新/删除，可调整过滤条件）
    new_records = batch_df.filter(col("_change_type") == "insert")
    # 移除CDF自带的元数据字段
    clean_records = new_records.drop("_change_type", "_commit_version", "_commit_timestamp")
    
    clean_records.write.format("jdbc")\
        .option("url", "jdbc:sqlserver://<你的SQL Server地址>.database.windows.net:1433;databaseName=<数据库名>")\
        .option("dbtable", "<目标表名>")\
        .option("user", "<用户名>")\
        .option("password", "<密码>")\
        .mode("append")\
        .save()

# 启动流式转发任务
silver_cdf_stream.writeStream.foreachBatch(write_cdf_to_sql).start()

关键注意点：

• 若需处理更新/删除操作，可调整_change_type的过滤条件（支持insert/update/delete）。
• 可通过startingVersion或startingTimestamp指定CDF的起始读取位置，避免重复转发历史数据。

额外优化建议 #

• 性能调优：根据数据量调整微批大小（通过trigger(processingTime='X seconds')或连续模式的continuous('Y seconds')参数）。
• 错误处理：在批量写入逻辑中添加异常捕获，记录失败日志，避免因单条数据失败导致整个微批重试。
• 连接池：使用spark.sql.jdbc.maxConnections参数设置JDBC连接池大小，避免频繁创建连接导致性能问题。

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