如何使用DataFlow读取复杂JSON格式并转换输出

一、读取复杂JSON数据源 #

1. 文件类JSON源：
   在Data Flow的源配置中，进入「JSON设置」面板：
   • 如果源JSON是数组格式（比如[{"a":1},{"a":2}]），将Document form设为Array of documents；
   • 如果是包含嵌套对象/数组的单文档JSON，勾选Allow complex types，确保Data Flow能识别嵌套层级的字段。
2. 字符串格式JSON：
   如果源数据是存储为字符串的JSON（比如数据库字段），用Derived Column转换，调用parseJson()函数解析：

   parseJson(json_string_column)
   

   解析后会生成可操作的复杂类型对象。

二、复杂JSON格式转换核心操作 #

针对嵌套结构重组、数组展开/聚合这类需求，用以下组合操作实现：

1. 展开嵌套数组

用Flatten转换处理数组字段：

• 选择要展开的数组列，设置展开模式：
  • Into rows：将数组的每个元素拆分为单独的行（适合一对多结构）；
  • Into columns：将数组元素按索引拆分为多列（适合固定长度的数组）；
• 多层嵌套数组需要多次使用Flatten，逐层展开。

2. 构建嵌套结构

用Derived Column转换，结合createMap()、createArray()函数构建目标嵌套格式：

• 构建嵌套对象：

  createMap("target_key1", source_column1, "target_key2", source_nested_column.field)
  

• 构建数组：

  createArray(source_columnA, source_columnB)
  

3. 聚合重组数组

如果需要将多行数据聚合成嵌套数组，用Aggregate转换：

• 选择分组键（比如订单ID）；
• 用collect()函数将指定字段聚合成数组：

  collect(derived_nested_object)
  

三、示例：从源格式到目标格式的转换流程 #

假设源JSON结构：

{
  "orderId": "O001",
  "customer": {
    "name": "张三",
    "contact": "138xxxxxxx"
  },
  "items": [
    {"productId": "P001", "quantity": 2},
    {"productId": "P002", "quantity": 1}
  ]
}

目标JSON结构：

{
  "order_info": {
    "id": "O001",
    "customer_name": "张三"
  },
  "products": [
    {"pid": "P001", "qty": 2},
    {"pid": "P002", "qty": 1}
  ]
}

转换步骤：

1. 解析与基础转换：用Derived Column创建order_info字段：

   createMap("id", orderId, "customer_name", customer.name)
   

2. 展开数组：用Flatten转换展开items数组，生成每行对应一个商品的数据集；
3. 构建商品对象：用Derived Column创建product字段：

   createMap("pid", items.productId, "qty", items.quantity)
   

4. 聚合数组：用Aggregate转换，按orderId分组，生成products数组：

   collect(product)
   

5. 筛选字段：用Select转换保留order_info和products字段，得到最终格式。

四、调试技巧 #

• 每完成一步转换，用Data Preview查看实时结果，快速定位结构错误；
• 处理空值时，用coalesce()或isNull()函数兜底，避免因字段缺失导致转换失败；
• 复杂结构拆分多步处理，不要一次性叠加多个转换逻辑，降低排查难度。

内容的提问来源于stack exchange，提问作者ADF-Learner