如何在ADF中忽略已处理的SharePoint文件且不移动至归档？

核心思路 #

无需移动文件至归档文件夹，通过持久化记录已处理文件的元数据（如文件名、创建/修改时间），每次管道运行时对比SharePoint目标文件夹内的文件，仅筛选并处理未被记录的新文件。

具体实现步骤 #

1. 搭建已处理文件记录存储 #

推荐用Azure Blob存储的JSON文件（低成本易维护）或Azure SQL Database（大文件量场景更高效）记录已处理文件信息：

• 若用Blob存储：在Azure存储账户创建容器（如processed-files-log），新建空JSON文件processed-files.json，初始内容为[]。

2. 管道核心流程设计 #

管道包含以下串联活动：

• Lookup活动：读取已处理文件记录
• Get Metadata活动：获取SharePoint目标文件夹的所有文件元数据
• Filter活动：筛选出未处理的新文件
• ForEach活动：遍历处理新文件
• Append Variable活动：更新已处理文件列表
• Copy活动：将更新后的列表写回记录存储

3. 各活动详细配置 #

（1）Lookup活动

• 数据源选择Azure Blob存储，指向processed-files.json
• 取消勾选「First row only」，确保读取所有已记录条目

（2）Get Metadata活动

• 数据源选择SharePoint Online，指定目标文件夹
• 在「Field list」中勾选Child items，获取所有文件的名称、创建时间、修改时间等元数据

（3）Filter活动

• Items：@activity('Get Metadata').output.childItems
• Condition：仅处理未记录的文件时用@not(contains(activity('Lookup').output.value, item().name))；若需处理修改过的旧文件，可对比修改时间：@greater(item().lastModified, first(filter(activity('Lookup').output.value, x equals(x.name, item().name))).lastModified)（需在记录中同时存储文件名和修改时间）

（4）ForEach活动

• Items：@activity('Filter').output.value
• 循环体内添加你的业务处理逻辑（如Copy活动转存文件、Data Flow数据转换等）

（5）Append Variable活动

• 先创建Array类型变量processedFiles
• 循环内每次处理完文件后，执行Append Variable：若仅记录文件名用@item().name，若需记录时间则用@createObject('name', item().name, 'lastModified', item().lastModified)

（6）Copy活动

• 源数据集选「Inline dataset」，类型为JSON，内容设为@variables('processedFiles')
• 目标数据集指向processed-files.json，设置「Copy behavior」为Overwrite，覆盖原记录文件

4. 触发器配置 #

• 定时触发器：按需求设置运行周期，每次自动筛选新文件处理
• 事件触发：借助Logic App或Azure Function监听SharePoint文件上传事件，触发ADF管道立即处理新文件

注意事项 #

• 确保ADF拥有SharePoint目标文件夹和记录存储的读写权限
• 大文件量场景建议用Azure SQL Database存储已处理记录，提升对比查询效率
• 可添加失败分支逻辑：处理失败的文件不写入已记录列表，下次运行自动重试

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