Azure Data Lake无Dataflow转换的格式转换与API加载管道开发需求

输入输出说明 #

• 输入文本（ADLS存储）：

BannerNumber    StoreGroupNumber    StoreGroupName  MemberNumber    MemberName  StoreNumbers
21                      83              SD           225             SD2        "5","10","20","30","40","41","42","43","44","93"

• 目标JSON格式：

{
    "BannerNumber": "21",
    "StoreGroupNumber": "83",
    "StoreGroupName": "SD",
    "StoreGroupMembers": [
        {
            "MemberId": "225",
            "MemberLabel": "SD2",
            "StoreNumbers": [
                {"StoreId": "5"},
                {"StoreId": "10"},
                {"StoreId": "20"},
                {"StoreId": "30"},
                {"StoreId": "40"},
                {"StoreId": "41"},
                {"StoreId": "42"},
                {"StoreId": "43"},
                {"StoreId": "44"},
                {"StoreId": "93"}
            ],
            "MemberAddedOn": "2023-10-23 14:09:29.0430000"
        }
    ]
}

注：MemberAddedOn为管道运行当天的日期时间

管道组件与步骤 #

1. 读取ADLS文本文件（Lookup活动） #

• 添加Lookup活动，配置数据源为ADLS中的目标文本文件，设置First row only为True（假设每个文件对应一条JSON数据）。
• 输出结果包含第一行数据的键值对，可通过@{activity('Lookup').output.firstRow.BannerNumber}这类表达式获取对应字段值。

2. 拆分StoreNumbers字段（Set Variable活动） #

• 创建字符串变量storeNumbersStr，赋值表达式：

@{replace(replace(activity('Lookup').output.firstRow.StoreNumbers, '"', ''), ',', '","')}

作用：先移除原有引号，再将逗号替换为带引号的分隔符，为后续转数组做准备

• 创建数组变量storeNumbersArray，赋值表达式：

@split(replace(activity('Lookup').output.firstRow.StoreNumbers, '"', ''), ',')

作用：直接去掉引号后按逗号拆分，得到门店ID的纯数组

3. 构建StoreNumbers数组对象（ForEach + Append Variable活动） #

• 创建数组变量formattedStores，用于存储格式化后的StoreId对象集合。
• 添加ForEach活动，迭代项设为@{variables('storeNumbersArray')}。
• 在ForEach内部添加Append Variable活动，向formattedStores追加对象：

{"StoreId": "@{item()}"}

4. 拼接完整JSON结构（Set Variable活动） #

• 创建字符串变量finalJson，用ADF表达式拼接目标JSON结构：

@{
    json(concat(
        '{',
            '"BannerNumber": "', activity('Lookup').output.firstRow.BannerNumber, '",',
            '"StoreGroupNumber": "', activity('Lookup').output.firstRow.StoreGroupNumber, '",',
            '"StoreGroupName": "', activity('Lookup').output.firstRow.StoreGroupName, '",',
            '"StoreGroupMembers": [',
                '{',
                    '"MemberId": "', activity('Lookup').output.firstRow.MemberNumber, '",',
                    '"MemberLabel": "', activity('Lookup').output.firstRow.MemberName, '",',
                    '"StoreNumbers": ', string(variables('formattedStores')), ',',
                    '"MemberAddedOn": "', formatDateTime(utcnow(), 'yyyy-MM-dd HH:mm:ss.fffffff'), '"',
                '}',
            ']',
        '}'
    ))
}

作用：用utcnow()获取当前时间，通过formatDateTime格式化要求的时间格式，最后用json()函数确保输出为合法JSON

5. 调用API加载数据（Web活动） #

• 添加Web活动，配置如下：
  • 请求方法：POST（根据API实际要求调整）
  • API地址：目标服务的URL
  • 请求正文：@{variables('finalJson')}
  • 按API要求配置认证方式（如API Key、OAuth等）

注意事项 #

• 若输入文件有多行数据，需修改Lookup活动为读取所有行，在外层增加ForEach活动循环处理每一行数据。
• 若文本文件使用固定分隔符（如制表符），需确保Lookup活动的数据源正确识别分隔符，避免字段取值错误。
• 时间格式可根据API要求调整formatDateTime的参数，如需本地时间可使用convertTimeZone函数转换时区。

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