如何不使用Dataflow Activity创建Azure Data Factory管道处理ADLS数据并对接API

方案一：纯ADF内置活动实现 #

1. 读取源数据 #

使用Lookup活动连接ADLS源数据集（如CSV/Parquet），读取目标数据记录。活动输出的value数组会包含单条（或批量）源数据，示例结构如下：

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

2. 拆分StoreNumbers为数组 #

创建字符串数组变量varStoreIds，用Set Variable活动执行拆分逻辑：

split(replace(activity('LookupSource').output.value[0].StoreNumbers, ' ', ''), ',')

该表达式先移除逗号后的空格，再按逗号拆分得到纯数字字符串数组。

3. 构造StoreNumbers嵌套对象数组 #

创建对象数组变量varStoreNumberObjects，用For Each活动遍历varStoreIds，循环内通过Append Variable活动生成嵌套对象：

{ "StoreId": item() }

4. 拼接完整目标JSON #

创建字符串变量varFinalJson，用Set Variable活动拼接符合要求的JSON结构，其中MemberAddedOn用utcNow()生成当前时间：

concat(
  '{',
  '"BannerNumber": "', activity('LookupSource').output.value[0].BannerNumber, '",',
  '"StoreGroupNumber": "', activity('LookupSource').output.value[0].StoreGroupNumber, '",',
  '"StoreGroupName": "', activity('LookupSource').output.value[0].StoreGroupName, '",',
  '"StoreGroupMembers": [',
    '{',
    '"MemberId": "', activity('LookupSource').output.value[0].MemberNumber, '",',
    '"MemberLabel": "', activity('LookupSource').output.value[0].MemberName, '",',
    '"StoreNumbers": ', string(variables('varStoreNumberObjects')), ',',
    '"MemberAddedOn": "', utcNow('yyyy-MM-dd HH:mm:ss.fffffff'), '"',
    '}',
  ']',
  '}'
)

若需批量处理多条记录，在外层增加For Each活动遍历Lookup返回的value数组，重复上述步骤后合并结果。

5. 输出结果 #

用Copy活动将varFinalJson通过内联JSON数据集写入ADLS；若目标为API，改用Web活动将varFinalJson作为请求体发送。

方案二：Azure Function辅助转换 #

若纯ADF步骤繁琐，可借助Azure Function简化逻辑：

1. ADF中用Lookup活动读取源数据，将整条记录作为参数传入Azure Function活动。
2. 在Azure Function中编写代码完成转换，示例Python核心逻辑：

import json
from datetime import datetime
import azure.functions as func

def main(req: func.HttpRequest) -> func.HttpResponse:
    req_body = req.get_json()
    source_data = req_body["sourceRecord"]
    
    # 拆分并生成StoreNumbers嵌套数组
    store_ids = [{"StoreId": s.strip()} for s in source_data["StoreNumbers"].split(',')]
    
    # 构造目标JSON
    target_json = {
        "BannerNumber": str(source_data["BannerNumber"]),
        "StoreGroupNumber": str(source_data["StoreGroupNumber"]),
        "StoreGroupName": source_data["StoreGroupName"],
        "StoreGroupMembers": [
            {
                "MemberId": str(source_data["MemberNumber"]),
                "MemberLabel": source_data["MemberName"],
                "StoreNumbers": store_ids,
                "MemberAddedOn": datetime.utcnow().strftime('%Y-%m-%d %H:%M:%S.%f')
            }
        ]
    }
    
    return func.HttpResponse(json.dumps(target_json), mimetype="application/json")

3. Azure Function返回构造好的JSON，ADF再将结果写入ADLS或调用目标API。

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