Azure Data Factory：跨服务器Upsert后获取并同步Identity Id

• 使用Azure Data Factory（ADF）的Copy Data活动，从CSV文件更新本地SQL Server数据库表，目标表包含int类型的Identity Id列，执行Upsert操作时自动生成该值，需获取此Id用于后续管道
• 限制：使用自托管Integration Runtime，无法使用数据流
• 补充需求：从源数据库加载约7000行数据到目标数据库，并将生成的Identity Id回写到源数据库；源、目标数据库位于不同服务器
• 尝试方向：计划使用带OUTPUT子句的MERGE语句执行Upsert并返回结果集，但Copy Data活动中MERGE的SELECT语句只能取自目标库，询问是否可通过Script活动实现

针对你的场景，完全可以用Script活动替代Copy Data活动实现需求，具体步骤如下：

1. 确保跨库访问权限 #

在目标SQL Server上配置对源SQL Server的访问：

• 方式一：创建链接服务器，让目标库可直接访问源库
• 方式二：确保目标SQL Server的账号具备源库的读取权限，通过四部分命名（[SourceServer].[SourceDB].[Schema].[Table]）访问源数据（需网络连通）

2. 编写带OUTPUT的MERGE脚本并处理回写 #

修改MERGE脚本，将Upsert的输出结果暂存到临时表，再通过关联字段将生成的Identity Id回写到源库。

完整脚本示例 #

-- 创建临时表存储MERGE输出结果
CREATE TABLE #MergeResults (
    ActionType NVARCHAR(10),
    TargetId INT,
    SourceSku NVARCHAR(50)
)

-- 执行MERGE并将输出插入临时表
MERGE Product AS target
USING (
    -- 从源库读取数据，替换为你的源库访问方式
    SELECT [epProductDescription], [epProductPrimaryReference]
    FROM [SourceServer].[SourceDB].[dbo].[epProduct]
    WHERE [epEndpointId] = '438E5150-8B7C-493C-9E79-AF4E990DEA04'
) AS source
ON target.[Sku] = source.[epProductPrimaryReference]
WHEN MATCHED THEN
    UPDATE SET 
        [Name] = source.[epProductDescription],
        [Sku] = source.[epProductPrimaryReference]
WHEN NOT MATCHED THEN
    INSERT ([Name], [Sku])
    VALUES (source.[epProductDescription], source.[epProductPrimaryReference])
-- 输出关键字段：操作类型、目标Id、源Sku（用于回写关联）
OUTPUT 
    $action,
    COALESCE(inserted.Id, updated.Id) AS TargetId,
    source.[epProductPrimaryReference] AS SourceSku
INTO #MergeResults;

-- 将生成的Identity Id回写到源库
UPDATE src
SET src.[epProductTargetId] = mr.TargetId
FROM [SourceServer].[SourceDB].[dbo].[epProduct] src
JOIN #MergeResults mr ON src.[epProductPrimaryReference] = mr.SourceSku
WHERE src.[epEndpointId] = '438E5150-8B7C-493C-9E79-AF4E990DEA04';

-- 若后续管道需要使用这些Id，可添加查询语句输出结果
-- SELECT TargetId, SourceSku FROM #MergeResults;

-- 清理临时表
DROP TABLE #MergeResults;

3. 在ADF中配置Script活动 #

• 选择目标SQL Server的连接（需具备源库的读取和更新权限）
• 将上述脚本粘贴到Script活动的「脚本内容」中
• 如果需要后续管道使用生成的Identity Id，可取消脚本中SELECT TargetId, SourceSku FROM #MergeResults;的注释，再通过Lookup活动读取该结果集，传递给后续活动

关键注意事项 #

• 若无法使用四部分命名，可改用OPENQUERY访问源库，示例：

  SELECT * FROM OPENQUERY([SourceLinkedServer], 
  'SELECT [epProductDescription], [epProductPrimaryReference] FROM [SourceDB].[dbo].[epProduct] WHERE [epEndpointId] = ''438E5150-8B7C-493C-9E79-AF4E990DEA04''')
  

• 7000行数据量较小，该方案性能完全满足需求，无明显延迟
• 确保执行脚本的SQL账号具备目标库的MERGE权限和源库的UPDATE权限

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