无需Dataflow的ADF管道：Azure DataLake数据聚合转JSON对接API

需求说明 #

将Azure Data Lake Storage（ADLS）中的源数据，按BannerNumber、StoreGroupNumber、StoreGroupName字段分组聚合，转换为指定嵌套JSON格式后写入ADLS（最终目标为API接口），禁止使用Dataflow转换功能。

输入数据 #

源数据示例（假设为结构化存储格式，如CSV/Parquet）：

BannerNumber    StoreGroupNumber    StoreGroupName  MemberNumber    MemberName  StoreNumbers
21                      83              SD           225             SD2        "5","10","20","30","40","41"
21                      83              SD           225             SD3        "6","10","20"
22                      84              SD           226             SD3        "43","44"

期望输出JSON格式 #

输出为多行JSON，每行对应一个分组的聚合结果：

第1行输出 #

{
    "BannerNumber": "21",
    "StoreGroupNumber": "83",
    "StoreGroupName": "SD",
    "StoreGroupMembers": [
        {
            "MemberId": "225",
            "MemberLabel": "SD2",
            "StoreNumbers": [
                {"StoreId": "5"},
                {"StoreId": "10"},
                {"StoreId": "20"},
                {"StoreId": "30"},
                {"StoreId": "40"},
                {"StoreId": "41"}
            ],
            "MemberAddedOn": "2024-05-20 15:30:00.0000000"
        },
        {
            "MemberId": "225",
            "MemberLabel": "SD3",
            "StoreNumbers": [
                {"StoreId": "6"},
                {"StoreId": "10"},
                {"StoreId": "20"}
            ],
            "MemberAddedOn": "2024-05-20 15:30:00.0000000"
        }
    ]
}

第2行输出 #

{
    "BannerNumber": "22",
    "StoreGroupNumber": "84",
    "StoreGroupName": "SD",
    "StoreGroupMembers": [
        {
            "MemberId": "226",
            "MemberLabel": "SD3",
            "StoreNumbers": [
                {"StoreId": "43"},
                {"StoreId": "44"}
            ],
            "MemberAddedOn": "2024-05-20 15:30:00.0000000"
        }
    ]
}

注：MemberAddedOn字段需自动生成为管道运行当天的时间戳

实现步骤（无Dataflow） #

步骤1：读取源数据 #

使用Lookup活动连接ADLS中的源数据集，获取全量数据。确保源数据集的格式配置正确（如CSV需指定分隔符、引号处理规则）。

步骤2：数据聚合与格式转换 #

由于禁用Dataflow，采用Azure Function活动处理核心逻辑：

1. 创建一个Azure Function（推荐C#/Python），接收Lookup活动返回的源数据数组。
2. 在Function中实现以下逻辑：
   • 按BannerNumber、StoreGroupNumber、StoreGroupName三个字段分组
   • 对每组内的成员数据，将StoreNumbers字段的字符串数组转换为{"StoreId": "xxx"}格式的对象数组
   • 为每个成员生成当前时间戳作为MemberAddedOn值
   • 组装成目标嵌套JSON结构，输出为多行JSON数组
3. 在ADF管道中调用该Azure Function，将Lookup的输出作为输入参数传入。

步骤3：写入ADLS目标位置 #

使用Copy Activity，将Azure Function返回的JSON数据写入ADLS指定路径：

• 源数据集选择Azure Function类型，指向步骤2创建的Function
• 目标数据集选择ADLS存储类型，配置为写入多行JSON格式
• 若后续需对接API，可在此步骤后添加Web活动，将生成的JSON作为请求体发送至目标API。

关键细节处理 #

• StoreNumbers字段解析：源数据中是带引号的字符串（如"5","10"），需在Azure Function中先去除引号、拆分字符串，再转换为目标数组格式
• 时间戳生成：在Function中使用当前系统时间生成符合格式的字符串（如C#中DateTime.UtcNow.ToString("yyyy-MM-dd HH:mm:ss.fffffff")）
• 分组逻辑：确保分组键的类型一致性（如将数字类型转换为字符串，匹配输出格式）

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