Azure Synapse Notebook中Python实现SCD Type 2报错求助

问题场景 #

在Azure Synapse Notebook中使用Python代码为维度表实现SCD Type2时，执行Merge操作遇到报错，需求是将现有表（existing_table）中surrogatekey的最大值加1，作为新插入记录的代理键值。

原代码如下：

(
existing_table.alias("old")
   .merge(
       new_table.alias("new"),
       "old.keycol = new.keycol"
   )
   .whenMatchedUpdate(condition = "old.CurrentRecord = true AND old.Rate <> new.rate"
       ,set = {"old.RecordEndDate" : "new.RecordStartDate"
              ,"old.CurrentRecord" : "false"
              }
       )
   .whenNotMatchedInsert(values = {"old.surrogatekey" : max("old.surrogatekey")+1
                                  ,"old.keycol" : "new.keycol"
                                  ,"old.Rate" : "new.Rate"
                                  ,"old.RecordStartDate" : "new.RecorStartDate"
                                  ,"old.CurrentRecord" : "true"
                                  }
       )
   .execute()
)  

报错信息 #

AnalysisException: cannot resolve old.surrogatekey in INSERT clause given colums new.keycol, new.Rate, new.RecordStartDate

问题原因 #

1. whenNotMatchedInsert无法引用old表字段：触发该分支时，old表无匹配记录，无法直接访问其字段，也不能在插入语句中直接计算max(old.surrogatekey)。
2. 拼写错误：new.RecorStartDate少写了一个'd'，应为new.RecordStartDate。
3. 字段前缀错误：whenNotMatchedInsert的values参数中，目标字段无需加old.前缀，直接用维度表列名即可。

修正方案 #

1. 预先计算最大代理键值：先查询existing_table的surrogatekey最大值，加1得到新记录的起始键；批量插入时用row_number()生成连续递增的键值。
2. 修正字段前缀与拼写错误：去掉values中字段的old.前缀，修正RecordStartDate的拼写。

修正后的完整代码 #

from pyspark.sql import functions as F
from pyspark.sql.window import Window

# 计算现有表最大代理键，表为空时默认设为0
max_surrogate_key = existing_table.select(F.coalesce(F.max("surrogatekey"), F.lit(0))).first()[0]

# 为新表添加自增的代理键列
new_table_with_sk = new_table.withColumn(
    "surrogatekey",
    F.row_number().over(Window.orderBy("keycol")) + max_surrogate_key
)

# 执行Merge操作
(
    existing_table.alias("old")
    .merge(
        new_table_with_sk.alias("new"),
        "old.keycol = new.keycol"
    )
    .whenMatchedUpdate(
        condition = "old.CurrentRecord = true AND old.Rate <> new.Rate",
        set = {
            "RecordEndDate": "new.RecordStartDate",
            "CurrentRecord": "false"
        }
    )
    .whenNotMatchedInsert(
        values = {
            "surrogatekey": "new.surrogatekey",
            "keycol": "new.keycol",
            "Rate": "new.Rate",
            "RecordStartDate": "new.RecordStartDate",
            "CurrentRecord": "true"
        }
    )
    .execute()
)

说明：

• 用coalesce处理空表场景，避免max()返回null导致后续计算出错。
• row_number()确保批量插入新记录时，代理键唯一且连续递增。
• Merge的set和values参数直接使用目标表列名，无需额外前缀。

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