在ADF数据流中封装JSON行至指定JSON结构的实现方法

需求说明：通过ODBC获取CSV数据后，已存储为Blob中的JSON行数据集（共2万多行），需要将其转换为包含Id（传入参数）、Name（自动生成UUID）、data（所有源JSON行组成的数组）的JSON结构，用于后续Web Activity调用API。

源JSON行示例：

{"A":12345,"B":"2023-12-31 00:00:00","C":5510,"D":20231231,"E":20240103,"F":20231231,"G":1577158,"H":63217,"I":null,"J":"Company123","K":"SupplierXYZ","L":"987","M":"Vendor ABC","N":"Invoice789","O":"2023-12-31T00:00:00","P":"2024-01-03T00:00:00","Q":"2023033344","R":"7559","S":"Dept123","T":"ModelXYZ","U":7179.0,"V":3,"W":"3769686.0","X":"TypeJ","Y":7179.0,"Z":null,"AA":"NOK","AB":"202312.0","AC":"EXTERNAL","AD":null,"AE":"Counterparty999","AF":"2023-12-31T00:00:00","AG":null,"AH":null,"AI":"66c6c4cc31107992a4079f3653a8a86930a2675d8663c5ab2ec4df23b0f6f3b0","AJ":"2024-02-13T12:58:02.597","AK":"2024-01-03T08:31:16.425292"}
{"A":54321,"B":"2024-01-15 08:00:00","C":7890,"D":20240115,"E":20240120,"F":20240115,"G":2468135,"H":951753,"I":null,"J":"Company456","K":"Supplier789","L":"123456789","M":"Vendor XYZ","N":"Invoice123","O":"2024-01-15T08:00:00","P":"2024-01-20T08:00:00","Q":"2024011515","R":"1234","S":"Dept456","T":"ModelABC","U":9876.0,"V":6,"W":"9876543.0","X":"TypeK","Y":9876.0,"Z":null,"AA":"USD","AB":"202401.0","AC":"INTERNAL","AD":null,"AE":"Counterparty888","AF":"2024-01-15T08:00:00","AG":null,"AH":null,"AI":"a1b2c3d4e5f6g7h8i9","AJ":"2024-02-15T08:00:00.123","AK":"2024-01-20T08:00:00.987654"}

期望输出结构：

{
    "Id": "传入的参数值",
    "Name": "自动生成的UUID",
    "data": [
        {"A":12345,...},
        {"A":54321,...}
    ]
}

具体实现步骤 #

1. 配置源数据集

   • 在ADF数据流中添加源，选择Blob存储中的JSON行数据集，确保格式设置为行分隔的JSON（Line-delimited JSON）。

2. 添加聚合转换（Aggregate）

   • 新增聚合转换，设置分组键为固定值（例如'group_all'），确保所有行被归入同一分组。
   • 创建新列data，使用表达式collect(@*)，该函数会将每一行的所有字段打包为JSON对象，并收集成完整数组。

3. 添加派生列转换（Derived Column）

   • 新增两个列：
     • Id：引用流水线参数，表达式示例：pipeline().parameters.Id（根据实际参数名调整）
     • Name：自动生成UUID，表达式：uuid()

4. 选择列转换（Select）

   • 仅保留Id、Name、data三个列，移除之前设置的分组键列。

5. 配置接收器（Sink）

   • 添加接收器，选择Blob存储的JSON数据集，格式设置为单个文档（Single document），确保输出为一个符合结构要求的JSON文件。

注意事项 #

• 聚合转换的分组键必须为固定值，保证所有行被聚合到同一组，否则collect(@*)只会收集分组内的行。
• 接收器必须设置为单个文档模式，避免输出多行JSON，确保结构符合API调用要求。

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