如何用PySpark实现Delta Table与SQL Table的关联同步？

一、核心技术术语 #

你描述的这种双向同步（插入/更新双向同步、删除操作仅允许SQL端发起）的表关联场景，核心术语是：

• 双向CDC同步（Bidirectional Change Data Capture Sync）：通过捕获两端表的变更数据，实现Delta表与外部SQL表的双向数据同步，同时通过规则限制删除操作的发起端。
• 外部Delta表（External Delta Table）：关联外部SQL数据源的Delta表，区别于默认创建的内部Delta表，可绑定外部数据源的同步规则。

二、PySpark实现关联同步的方法 #

你用Spark SQL创建的Delta表能实现关联，是因为DDL语句中指定了外部数据源的连接参数与同步规则；而直接使用df.write.format("delta").saveAsTable()默认创建的是内部Delta表，未绑定外部SQL源。以下是两种PySpark实现方式：

方式1：PySpark执行Spark SQL DDL（与现有Spark SQL逻辑对齐） #

直接通过spark.sql()执行DDL，创建关联外部SQL表的Delta表，同步规则通过表属性配置：

# 替换为你的实际配置
jdbc_url = "jdbc:sqlserver://<sql-host>:<port>;databaseName=<db-name>"
jdbc_user = "<username>"
jdbc_password = "<password>"
sql_table = "<sql-table-name>"
delta_table_path = "/dbfs/path/to/delta/table"
delta_table_name = "<delta-table-name>"

# 创建关联Delta表
spark.sql(f"""
CREATE TABLE {delta_table_name}
USING delta
LOCATION '{delta_table_path}'
TBLPROPERTIES (
  'delta.sync.jdbc.url' = '{jdbc_url}',
  'delta.sync.jdbc.user' = '{jdbc_user}',
  'delta.sync.jdbc.password' = '{jdbc_password}',
  'delta.sync.jdbc.table' = '{sql_table}',
  'delta.sync.allow.delete' = 'false' -- 禁止Delta端删除操作同步到SQL表
)
AS SELECT * FROM jdbc.{sql_table}
""")

注：需确保集群已安装对应SQL数据库的JDBC驱动（如SQL Server的com.microsoft.sqlserver:mssql-jdbc）。

方式2：纯PySpark DataFrame API实现 #

通过DataFrameWriter配置JDBC参数与Delta表属性，创建关联外部源的Delta表：

# 读取外部SQL表数据
df = spark.read \
    .format("jdbc") \
    .option("url", jdbc_url) \
    .option("dbtable", sql_table) \
    .option("user", jdbc_user) \
    .option("password", jdbc_password) \
    .load()

# 写入关联Delta表
df.write \
    .format("delta") \
    .option("path", delta_table_path) \
    .option("delta.sync.jdbc.url", jdbc_url) \
    .option("delta.sync.jdbc.user", jdbc_user) \
    .option("delta.sync.jdbc.password", jdbc_password) \
    .option("delta.sync.jdbc.table", sql_table) \
    .option("delta.sync.allow.delete", "false") \
    .saveAsTable(delta_table_name)

补充：双向同步规则配置 #

要实现完整的镜像逻辑，还需完成以下配置：

1. 为外部SQL表开启CDC（如SQL Server的Change Tracking、MySQL的Binlog），用于捕获SQL端的变更同步到Delta表。
2. 为Delta表开启CDC：执行ALTER TABLE <delta-table-name> SET TBLPROPERTIES (delta.enableChangeDataCapture = true)。
3. 创建两个流式作业：
   • 读取SQL表的CDC数据，写入Delta表。
   • 读取Delta表的CDC数据，过滤删除操作后写入SQL表，确保仅SQL端的删除生效。

三、技术文档参考 #

在Databricks官方文档中，可查找以下内容：

• Delta Lake CDC功能：搜索「Delta Lake Change Data Capture」，了解Delta表CDC的开启与使用。
• 外部Delta表配置：搜索「Delta Lake External Tables」，查看外部表的创建与关联规则。
• JDBC数据源连接：搜索「Databricks JDBC Connectivity」，获取外部SQL数据库的连接配置细节。
• Delta Live Tables同步：若使用DLT实现流式同步，搜索「Delta Live Tables CDC」查看配置方法。

内容的提问来源于stack exchange，提问作者Sergio Di Bella