如何在Azure Data Factory中合并Filter与ForEach内Lookup的输出？

具体实现步骤 #

1. 收集循环内Lookup的输出 #

• 先在流水线全局变量中创建一个数组类型的变量，比如命名为loopLookupResults，初始值设为[]（空数组）
• 在ForEach循环内部，每次Lookup活动执行完成后，添加一个Set Variable活动，将当前Lookup的输出追加到全局数组变量中，赋值表达式用：

  @concat(variables('loopLookupResults'), activity('你的Lookup活动名称').output.value)
  

  如果需要去重，改用union函数：

  @union(variables('loopLookupResults'), activity('你的Lookup活动名称').output.value)
  

2. 合并两类数组 #

ForEach循环执行完毕后，添加另一个Set Variable活动，创建数组变量combinedResults，用表达式合并Filter活动的输出和循环积累的数组：

• 不需要去重时用concat：

  @concat(activity('你的Filter活动名称').output.value, variables('loopLookupResults'))
  

• 需要去重用union：

  @union(activity('你的Filter活动名称').output.value, variables('loopLookupResults'))
  

3. 传递给Notebook活动 #

在Notebook活动的Base Parameters配置项中，新增一个参数（比如combined_dataset），参数值设置为：

@variables('combinedResults')

在Notebook内部（以PySpark为例），可以通过以下代码获取并解析数据：

import json
combined_data = json.loads(dbutils.widgets.get("combined_dataset"))
# 后续处理逻辑

关键注意点 #

• 全局数组变量必须初始化为空数组[]，否则合并时会出现格式错误
• 若Filter和Lookup的输出结构不一致，需在对应活动后添加Select活动统一字段结构
• 如果数组元素过多，超出ADF变量大小限制，可考虑将临时数据写入ADLS Gen2存储，再让Notebook读取该路径下的文件

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