如何在Azure Data Factory数据流中动态拆分扁平化XML？

针对你这种无法预先知晓invoiceNumber、需要动态拆分输出的场景，以下是几种实用的实现方案：

核心思路 #

先将扁平化后的同invoiceNumber数据分组，再为每个分组生成独立的JSON文件，关键是实现动态文件名和输出路径的配置。

方案一：基于ETL数据流工具（如SSIS/ADF）实现 #

1. 分组聚合数据

在数据流中添加聚合转换，按invoiceNumber分组，将该编号下的所有行合并为一个数组字段（比如命名为InvoiceData）。这样每个输出行就对应一个唯一的invoiceNumber及其完整数据集合。

2. 动态输出到JSON文件

• 如果用SSIS：

  • 使用脚本组件作为目标，在C#/VB脚本中读取每行的invoiceNumber和InvoiceData字段，将数组序列化为JSON字符串，然后写入路径为[输出目录]\Invoice_{invoiceNumber}.json的文件中。注意处理文件覆盖、目录权限问题。
  • 或者先通过SQL任务/数据流提取所有唯一的invoiceNumber到变量，再用Foreach循环容器遍历每个编号，在循环内的数据流中过滤出对应数据，输出到动态命名的JSON文件。

• 如果用Azure Data Factory（ADF）：

  • 用派生列生成动态文件名：concat('Invoice_', invoiceNumber, '.json')。
  • 在接收器中设置分区输出，将分区键指定为invoiceNumber，并配置文件名包含分区键值，即可自动为每个编号生成独立文件。
  • 也可以先通过Lookup活动获取所有唯一invoiceNumber，再用Foreach活动遍历每个编号，在循环内的数据流过滤数据后输出到动态命名文件。

方案二：脚本式处理（Python/PowerShell） #

如果可以脱离可视化ETL工具，用脚本处理更灵活：

• 读取扁平化后的数据集（比如CSV、DataFrame），按invoiceNumber分组。
• 遍历每个分组，将数据转换为JSON格式，写入以Invoice_{invoiceNumber}.json命名的文件。
• 示例（Python）：

  import pandas as pd
  import json
  
  # 读取扁平化后的数据
  df = pd.read_csv("flattened_invoices.csv")
  
  # 按invoiceNumber分组并输出
  for invoice_num, group in df.groupby("invoiceNumber"):
      # 转换为JSON格式
      json_data = group.to_json(orient="records", indent=2)
      # 写入文件
      with open(f"Invoice_{invoice_num}.json", "w", encoding="utf-8") as f:
          f.write(json_data)
  

注意事项 #

• 确保输出目录存在，必要时在代码/工具中添加自动创建目录的逻辑。
• 若存在重复的invoiceNumber（同一批数据内），需确保分组逻辑正确合并所有对应行。
• 处理大数据量时，优先采用“排序后逐行写入对应文件”的方式，避免内存过载。

内容的提问来源于stack exchange，提问作者Jose Angel Martinez Soler