Synapse中Delta表变更检测：缺失table_changes函数的替代方案咨询

问题背景 #

需要构建从Silver到Gold表的增量数据迁移流程，仅处理Silver表的变更记录。Delta Lake v2+的Change Data Feed(CDF)功能匹配需求，但Synapse暂未实现用于查询变更记录的table_changes表值函数，执行相关SQL时会报错：

Error: could not resolve table_changes to a table-valued function; line 2 pos 14
org.apache.spark.sql.catalyst.analysis.package$AnalysisErrorAt.failAnalysis(package.scala:42)
org.apache.spark.sql.catalyst.analysis.ResolveTableValuedFunctions$$anonfun$apply$1.$anonfun$applyOrElse$2(ResolveTableValuedFunctions.scala:37)

复现步骤（Synapse Spark Notebook） #

1. 创建基础Silver Delta表

countries = [("USA", 10000, 20000), ("India", 1000, 1500), ("UK", 7000, 10000), ("Canada", 500, 700) ]
columns = ["Country","NumVaccinated","AvailableDoses"]
spark.createDataFrame(data=countries, schema = columns).write.format("delta").mode("overwrite").saveAsTable("silverTable")

2. 查询新创建的表

SELECT * FROM silverTable

3. 启用变更数据馈送设置

ALTER TABLE silverTable SET TBLPROPERTIES (delta.enableChangeDataFeed = true)

4. 添加变更数据用于提取

new_countries = [("Australia", 100, 3000)]
spark.createDataFrame(data=new_countries, schema = columns).write.format("delta").mode("append").saveAsTable("silverTable")

5. 尝试查询变更记录（报错）

SELECT * FROM table_changes('silverTable', 2, 5) order by _commit_timestamp

替代方案 #

方案1：流式读取CDF（适合持续增量同步）

利用Spark Streaming读取CDF，自动跟踪已处理的版本，适合持续运行的增量迁移任务：

# 读取Silver表的变更数据流，从版本2开始
changes_stream = spark.readStream \
    .format("delta") \
    .option("readChangeFeed", "true") \
    .option("startingVersion", 2) \
    .table("silverTable")

# 将变更数据写入Gold表，通过checkpoint记录处理进度
write_query = changes_stream.writeStream \
    .format("delta") \
    .option("checkpointLocation", "/synapse/workspaces/your-workspace/checkpoints/silver-to-gold") \
    .table("goldTable")

write_query.awaitTermination()

• 说明：checkpointLocation需指定Synapse可访问的路径，任务重启后会从上次中断处继续处理；可通过_change_type字段过滤所需的变更类型（如只保留insert和update_postimage）。

方案2：批处理读取指定版本范围的CDF（适合定时增量任务）

通过批处理方式读取特定版本区间的变更数据，适合定时执行的增量迁移：

# 获取Silver表的最新版本
latest_version = spark.sql("DESCRIBE HISTORY silverTable") \
    .select("version") \
    .orderBy("version", ascending=False) \
    .first()[0]

# 读取版本2到最新版本的变更数据
changes_df = spark.read \
    .format("delta") \
    .option("readChangeFeed", "true") \
    .option("startingVersion", 2) \
    .option("endingVersion", latest_version) \
    .table("silverTable")

# 过滤业务需要的变更（示例：保留插入和更新后的数据）
valid_changes = changes_df.filter("_change_type IN ('insert', 'update_postimage')")

# 将变更写入Gold表
valid_changes.write.format("delta").mode("append").saveAsTable("goldTable")

• 说明：DESCRIBE HISTORY可查看Delta表的所有版本记录；_change_type包含四种类型：insert（插入）、update_preimage（更新前数据）、update_postimage（更新后数据）、delete（删除）。

方案3：版本对比法（适合未启用CDF或简单场景）

如果未启用CDF，或需要更灵活的差异对比，可通过读取两个版本的表进行主键匹配，找出变更：

# 读取版本2的Silver表
version_2_df = spark.read.format("delta").option("versionAsOf", 2).table("silverTable")
# 读取最新版本的Silver表
latest_df = spark.read.table("silverTable")

# 找出新增/更新的记录（以Country作为主键）
new_updated = latest_df.join(version_2_df, on="Country", how="left_anti")
# 找出删除的记录
deleted = version_2_df.join(latest_df, on="Country", how="left_anti")

# 标记变更类型并合并数据
final_changes = new_updated.withColumn("_change_type", lit("insert/update")) \
    .union(deleted.withColumn("_change_type", lit("delete")))

# 写入Gold表
final_changes.write.format("delta").mode("append").saveAsTable("goldTable")

• 说明：该方法依赖主键，无法区分插入和更新，若需区分需额外对比字段值；适合主键明确、变更逻辑简单的场景。

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