如何在Azure Data Factory中批量推送SQL Server数据至POST API

步骤1：配置SQL Server数据源 #

• 创建连接目标SQL Server的链接服务，确保拥有目标表的读取权限。
• 创建对应数据集，指向待同步的表，映射id、field1、field2三个字段。

步骤2：拆分数据为100条/批次 #

由于API单请求上限为100条，需将65023条数据拆分为651批（650批满100条，最后1批23条），可通过两种方式实现：

方式A：Lookup活动+循环（适合中等数据量） #

1. 添加Lookup活动，执行查询获取总记录数：

   SELECT COUNT(*) AS total FROM [你的表名]
   

2. 创建三个整数变量：totalRecords（赋值为Lookup返回的总记录数）、currentPage（初始值1）、pageSize（固定为100）。
3. 添加Until循环活动，终止条件设为：

   @greaterOrEquals(variables('currentPage'), ceiling(div(variables('totalRecords'), variables('pageSize'))))
   

4. 在循环内部添加Lookup活动，执行分页查询获取当前批次数据：

   SELECT id, field1, field2
   FROM (
       SELECT *, ROW_NUMBER() OVER(ORDER BY id) AS row_num
       FROM [你的表名]
   ) t
   WHERE row_num BETWEEN @{mul(sub(variables('currentPage'),1), variables('pageSize')) +1} AND @{mul(variables('currentPage'), variables('pageSize'))}
   

   注：用id排序可保证数据不重复不遗漏，也可替换为其他有序字段。

方式B：Copy活动+内置分页（更高效，适合大数据量） #

1. 创建REST类型数据集，设置请求方法为POST，端点填写https://api.local/batch/update。
2. 在Copy活动的源配置中选择SQL Server数据集，启用分页功能，选择“分页查询”，模板填写：

   SELECT id, field1, field2 FROM [你的表名] ORDER BY id OFFSET @{offset} ROWS FETCH NEXT @{pageSize} ROWS ONLY
   

   ADF会自动计算offset和pageSize参数，实现批量拆分。

步骤3：构造符合要求的API请求体 #

对应方式A（Lookup+循环） #

在循环内添加Set Variable活动，用ADF表达式转换数据结构：

1. 先将Lookup返回的单条记录转换成API要求的格式：

   @map(activity('Lookup当前批次').output.value, 
       item() => json(concat('{"id": "', item().id, '", "properties": {"field1": "', item().field1, '", "field2": "', item().field2, '"}}'))
   )
   

2. 再将数组包装成完整请求体：

   @concat('{"inputs": ', string(variables('currentBatchData')), '}')
   

对应方式B（Copy活动） #

在Copy活动的映射中通过派生列构造嵌套结构：

1. 派生列properties，表达式为：

   @json(concat('{"field1": "', field1, '", "field2": "', field2, '"}'))
   

2. 派生列item，表达式为：

   @json(concat('{"id": "', id, '", "properties": ', properties, '}'))
   

3. 最后将所有item聚合为数组，Copy活动会自动将其包装为{"inputs": [...]}格式的请求体。

步骤4：发送POST请求 #

对应方式A（Lookup+循环） #

在循环内部添加Web活动：

• URL填写https://api.local/batch/update
• 请求方法选择POST
• 请求体选择动态内容，填入之前构造好的请求体变量
• 设置请求头Content-Type: application/json

对应方式B（Copy活动） #

直接在Copy活动的目标配置中选择已创建的REST数据集，ADF会自动按批次发送请求。

步骤5：错误处理与批次递进 #

• 添加捕获活动，捕获请求失败的批次，将失败的批次号、数据内容记录至存储账户或日志表，便于后续重试。
• 若用方式A，在循环末尾添加Set Variable活动，将currentPage递增1，确保批次顺序执行。

内容的提问来源于stack exchange，提问作者Nestor Yara Gonzalez