ADF中如何用Data Flow构建批量REST调用的指定格式聚合JSON

你可以通过Data Flow内置的窗口、聚合转换组合实现需求，无需编写自定义代码，具体操作步骤如下：

1. 配置源转换 #

• 新增源转换，对接存储客户数据的数据源，确认架构正确识别CustomerId、Name、Country三个字段，数据类型匹配预期。

2. 通过窗口转换分配批次ID #

• 在源转换后新增窗口转换：
  • 「分区」选项卡不选择任何分区字段，所有数据划入同一个全局窗口
  • 「排序」选项卡选择按CustomerId升序排序，保证批次顺序和原始数据顺序一致
  • 新增窗口计算列：
    • 列名：BatchId
    • 表达式：ceil(rowNumber() / 2)

    表达式里的除数2就是单批次的最大记录数，后续需要调整批次大小直接修改这个值即可，例如每批100条就写ceil(rowNumber() / 100)

• 执行完这一步，示例数据中CustomerId为1、2的记录BatchId为1，CustomerId为3的记录BatchId为2，和预期批次划分完全一致。

3. 派生列调整字段名匹配API规范 #

• 窗口转换后新增派生列转换：
  • 新增两个字段：
    • name：值取原字段Name
    • country：值取原字段Country
  • 转换完成后删除冗余的CustomerId、Name、Country字段，仅保留BatchId、name、country三个字段。

4. 聚合生成批次JSON数组结构 #

• 派生列转换后新增聚合转换：
  • 「分组依据」选项卡选择BatchId作为分组字段
  • 「聚合」选项卡新增聚合列：
    • 列名：customers
    • 表达式：collect(@(name = name, country = country))

  collect()函数会将同组内的所有行聚合成数组，@()语法用于构造每个数组元素的JSON对象结构，完全匹配接口要求的字段格式。

• 这一步输出的结果会按BatchId拆分为独立行，每行对应一个批次，customers列即为该批次下的客户对象数组。

5. 配置REST接收器发送请求 #

• 聚合转换后新增接收器转换，对接你的REST API数据集：
  • 请求方法选择POST
  • 写入行为设置为逐行写入，确保每个BatchId对应的记录单独作为一个请求体发送
  • 请求体映射直接关联customers列，最终发出的请求JSON结构和你给出的示例完全一致。

结果校验 #

你可以通过Data Flow的数据预览功能查看每一步输出：示例数据经过聚合后会生成2条记录，第一条对应批次1的请求体，第二条对应批次2的请求体，结构完全符合接口要求。

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