如何在Azure Data Factory中拆分API数组至Blob存储并关联主记录ID

由于Copy Activity无法直接处理嵌套数组的拆分与扁平化，需要使用**数据流（Data Flow）**来实现将API响应中的tags和riskCauses数组拆分到独立Blob文件，且每条数据携带主记录ID的需求。以下是具体实现步骤：

一、配置数据源 #

1. API数据源：创建链接服务连接目标API，配置请求参数（如分页、认证），确保能正确获取包含嵌套数组的响应数据。
2. Blob存储数据源：创建链接服务指向目标Blob存储账户，设置好用于存储拆分后文件的容器路径。

二、构建数据流 #

1. 源节点配置 #

• 选择API作为数据源，在JSON 设置中指定JSON路径定位到主记录数组：$[0].records（匹配示例响应中的结构），确保能读取到每条主记录的完整数据（包括id、tags、riskCauses等字段）。

2. 拆分tags数组分支 #

• 添加**扁平化（Flatten）**节点：
  • 在展开设置中，选择tags作为要展开的数组列，设置展开方式为按数组展开。
  • 保留主记录的id字段（建议重命名为main_record_id），同时选择展开后的tags.id和tags.name字段。
• 添加**选择（Select）**节点：
  • 仅保留main_record_id、tags.id（重命名为tag_id）、tags.name（重命名为tag_name）三个字段，剔除无关数据。
• 添加**输出（Sink）**节点：
  • 选择Blob存储数据源，设置输出格式为JSON/CSV，指定文件路径（如container/tags/），确保每个输出文件包含拆分后的tags数据。

3. 拆分riskCauses数组分支 #

• 在源节点后添加**分支（Branch）**节点，创建第二个数据流分支：
  • 添加**扁平化（Flatten）**节点：选择riskCauses作为要展开的数组列，同样保留主记录的main_record_id，以及riskCauses.id、riskCauses.name、riskCauses.status字段。
• 添加**选择（Select）**节点：
  • 重命名字段为main_record_id、risk_cause_id、risk_cause_name、risk_cause_status，剔除无关字段。
• 添加**输出（Sink）**节点：
  • 选择Blob存储数据源，设置独立的输出路径（如container/risk_causes/），存储拆分后的riskCauses数据。

三、集成到管道 #

创建新管道，添加数据流活动，选择上述构建好的数据流，替代原有的Copy Activity。执行管道后，Blob存储中会生成两个独立的文件集合，分别存储tags和riskCauses数据，每条记录均携带主记录ID，可直接用于数据仓库表的关联。

示例输出 #

• Tags文件内容：

[
  {"main_record_id":100106,"tag_id":10175,"tag_name":"Test Tag 1"},
  {"main_record_id":100106,"tag_id":10160,"tag_name":"Test Tag 2"},
  {"main_record_id":100106,"tag_id":10163,"tag_name":"Test Tag 3"}
]

• RiskCauses文件内容：

[
  {"main_record_id":100106,"risk_cause_id":100009,"risk_cause_name":"Test Cause 1","risk_cause_status":0},
  {"main_record_id":100106,"risk_cause_id":100013,"risk_cause_name":"Test Cause 2","risk_cause_status":0},
  {"main_record_id":100106,"risk_cause_id":100066,"risk_cause_name":"Test Cause 3","risk_cause_status":0},
  {"main_record_id":100106,"risk_cause_id":100008,"risk_cause_name":"Test Cause 4","risk_cause_status":0}
]

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