如何使用Azure Data Factory将指定JSON对象转换为目标CSV格式

给定输入JSON：

{"IDpol": [406.0, 407.0, 412.0], "ClaimNb": [1, 1, 1], "Exposure": [0.77, 0.07, 0.84], "Area": ["C", "C", "E"], "VehPower": [9, 9, 7], "VehAge": [0, 0, 3], "DrivAge": [29, 29, 40], "BonusMalus": [72, 72, 54], "VehBrand": ["B12", "B12", "B12"], "VehGas": ["Regular", "Regular", "Regular"], "Density": [360, 360, 5746], "Region": ["R91", "R91", "R11"]}

目标CSV格式：

IDpol,ClaimNb,Exposure,Area,VehPower,VehAge,DrivAge,BonusMalus,VehBrand,VehGas,Density,Region
406.0 ,   1   ,  0.77  , C  ,   9    ,   0   ,   29  , 72      ,  B12   ,Regular,360   , R91
407.0 ,   1   ,  0.07  , C  ,    9   ,   0   ,   29  , 72      ,  B12   ,Regular,360   ,R91
412.0 ,   1   ,  0.84  , E  ,    7   ,   3   , 40    , 54      , B12    , Regular,5746, R11

操作步骤 #

1. 创建JSON数据源

   • 在Azure Data Factory中添加存储链接服务（如Blob Storage、ADLS Gen2），指向JSON文件所在存储位置。
   • 创建JSON格式数据集，关联上述链接服务，指定文件路径，设置读取模式为Single document。

2. 新建数据流

   • 创建数据流，添加源组件并选择刚才的JSON数据集。

3. 逆透视（Unpivot）转换

   • 添加Unpivot组件连接源输出：
     • 取消勾选"Use current columns"，选择所有顶级列（IDpol、ClaimNb等）作为逆透视列。
     • 设置columnName为Pivot key（存储原列名），columnValue为Value column（存储原列对应的数组）。
     • 转换后数据变为两列：每行对应一个原JSON键值对。

4. 展平（Flatten）转换

   • 添加Flatten组件连接Unpivot输出：
     • 选择columnValue作为展平的数组列。
     • 勾选"Generate index column"并命名为rowIndex（标记数组元素的位置索引）。
     • 转换后每行对应一个数组元素，包含columnName、columnValue、rowIndex三列。

5. 透视（Pivot）转换

   • 添加Pivot组件连接Flatten输出：
     • 设置rowIndex为Group by列（按索引分组，将同索引元素组合为一行）。
     • 设置columnName为Pivot key（用原列名作为新列名）。
     • 选择First作为聚合方式，聚合列选columnValue（每个索引+列名组合唯一，取首个值即可）。
     • 转换后数据结构与目标CSV的行结构一致。

6. 导出到CSV

   • 添加**接收器（Sink）**组件连接Pivot输出：
     • 创建CSV格式数据集，关联存储链接服务并指定输出路径。
     • 勾选"Include header"生成列标题行；如需匹配目标的空格格式，可在Pivot后添加派生列，对每个字段拼接空格（如concat(IDpol, ' ')）后再输出。
     • 在Mapping选项卡中调整列顺序，确保与目标一致。

7. 运行管道

   • 创建管道并添加Execute Data Flow活动，关联上述数据流。
   • 触发管道运行，生成符合要求的CSV文件。

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