如何在ADF中创建动态源批量处理文件并生成对应输出文件

步骤1：获取目标目录的文件列表 #

• 新建管道，添加Get Metadata活动，关联你的存储账户（Blob Storage/ADLS Gen2等），指定要处理的根目录路径。
• 在活动的字段设置中，勾选Child items，用于获取目录下所有文件的元数据；如果需要递归处理子目录，勾选Recurse选项。

步骤2：遍历每个文件 #

• 在Get Metadata活动后添加For Each活动，将Items属性设置为：

  @activity('Get Metadata活动名称').output.childItems
  

• 开启Sequential模式（强制顺序执行），确保每个文件逐个被处理。

步骤3：配置数据流的动态源 #

• 打开现有数据流，修改源数据集：
  1. 给源数据集添加字符串类型参数sourceFileName。
  2. 将数据集的文件路径设置为：

     @concat('你的源目录路径/', dataset().sourceFileName)
     

• 在For Each活动内添加Execute Data Flow活动，关联你的数据流。
• 在Execute Data Flow的参数页签，将数据流的sourceFileName参数赋值为：

  @item().name
  

  （item().name对应For Each遍历到的当前文件名）

步骤4：配置动态接收器（匹配源文件名输出） #

• 修改接收器数据集：
  1. 添加字符串类型参数targetFileName。
  2. 设置接收器的文件路径，比如要保留原文件名和后缀，直接写：

     @concat('你的输出目录/', dataset().targetFileName)
     

     如果需要修改后缀，用replace函数，例如：

     @concat('你的输出目录/', replace(dataset().targetFileName, '.csv', '.processed.csv'))
     

  3. 确保接收器的File pattern设置为Single file，保证每个源文件生成独立的输出文件。
• 回到Execute Data Flow活动，给接收器数据集的targetFileName参数赋值为：

  @item().name
  

额外注意事项 #

• 如果只需要处理特定后缀的文件，可在For Each的Items中添加过滤逻辑，例如只处理CSV文件：

  @filter(activity('Get Metadata活动名称').output.childItems, contains(item().name, '.csv'))
  

• 测试阶段先用少量文件验证流程，避免批量执行时出现大范围错误。
• 可在For Each内添加Set Variable或Web活动记录每个文件的处理状态，方便后续排查问题。

内容的提问来源于stack exchange，提问作者Luis Gustavo Martinez Osorio