ADF Copy Data使用Upsert时identity列空值报错解决方案咨询

问题背景 #

• 首次发帖咨询，如有表述不周还请谅解
• 业务场景：在Azure Data Factory管道中使用Copy Data活动执行数据同步，源数据为.parquet格式文件，目标为数据库表，原计划在Sink端采用Upsert模式写入
• 目标表规则：主键为ID字段，配置为步长1的自增identity列；配置Upsert时指定其他业务列作为匹配键，字段映射阶段已移除ID列的映射关系
• 故障现象：管道运行触发「无法向ID列插入NULL值」报错；测试验证全量Insert模式下不映射ID列可正常写入，仅切换为Upsert模式时执行失败
• 官方结论：已向微软支持中心反馈该问题，确认属于Upsert逻辑的已知Bug，官方给出的临时解决方案为通过自定义存储过程结合Merge语句实现Upsert逻辑

现有配置信息 #

源端配置 #

• 源数据集：data.parquet
• 文件路径类型：通配符文件路径
• 递归读取：已启用

Sink端配置 #

• Sink数据集：data_table
• 写入行为：Insert（计划修改为存储过程写入模式）
• Bulk Insert表锁：否
• 表选项：无
• 复制前脚本：delete from db.targettable
• 其余配置：均为空或未勾选

核心诉求 #

实现逻辑为：源数据与目标表的指定业务键匹配时执行更新，无匹配记录时执行插入；存储过程支持自定义Upsert匹配键列。因无存储过程编写经验，需要可直接适配场景的存储过程写法。

目前自行编写的存储过程逻辑存在错误，参考代码如下：

CREATE PROCEDURE [db].[prc_LoadData]
    @column1 NVARCHAR(19),
    @column2 NVARCHAR(10),
    @column3 NVARCHAR(10),
    @column4 DATE,
    @column5 DATE 
AS
BEGIN
    Select * from db.targettable where column1 = @column1,
    Select * from db.targettable where column2 = @column2,
    Select * from db.targettable where column3 = @column3,
    Select * from db.targettable where column4 = @column4,
    Select * from db.targettable where column5 = @column5
END

实现方案 #

ADF Copy活动使用存储过程作为Sink写入时，会将每行源数据的映射字段作为存储过程入参传入，直接在存储过程中用MERGE语句实现匹配逻辑即可，无需编写单独的SELECT查询语句。

可用存储过程代码 #

以下代码默认以column1作为匹配业务键，可根据实际业务需求调整匹配列、更新列：

CREATE PROCEDURE [db].[prc_LoadData]
    @column1 NVARCHAR(19),
    @column2 NVARCHAR(10),
    @column3 NVARCHAR(10),
    @column4 DATE,
    @column5 DATE 
AS
BEGIN
    -- 关闭多余行数返回，提升写入兼容性与性能
    SET NOCOUNT ON;

    MERGE [db].[targettable] AS Target
    USING (
        SELECT 
            @column1 AS column1,
            @column2 AS column2,
            @column3 AS column3,
            @column4 AS column4,
            @column5 AS column5
    ) AS Source
    -- 匹配规则：可通过AND拼接多列实现联合业务键匹配
    ON Target.column1 = Source.column1
    -- 匹配到存量记录时执行更新，无需操作自增ID列
    WHEN MATCHED THEN
        UPDATE SET
            Target.column2 = Source.column2,
            Target.column3 = Source.column3,
            Target.column4 = Source.column4,
            Target.column5 = Source.column5
    -- 未匹配到记录时执行插入，不传入ID列，由数据库自动生成自增值
    WHEN NOT MATCHED THEN
        INSERT (column1, column2, column3, column4, column5)
        VALUES (Source.column1, Source.column2, Source.column3, Source.column4, Source.column5);
END

配置注意事项 #

• ADF侧Sink配置需调整：将写入行为从Insert修改为存储过程模式，选择上述创建的[db].[prc_LoadData]，字段映射仅保留业务列，不要映射自增ID列
• 多列匹配调整：如果需要使用多列作为Upsert匹配键，直接在ON子句后追加条件即可，例如同时匹配column1和column2时，条件写为ON Target.column1 = Source.column1 AND Target.column2 = Source.column2
• 复制前脚本评估：现有配置中的delete from db.targettable为全量清空目标表逻辑，全量同步场景可保留，增量同步场景需删除该脚本，避免每次同步清空历史数据
• 存储过程中插入和更新语句均未操作ID自增列，不会触发NULL值插入报错

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