基于文件夹结构与日期条件的Azure Data Factory文件拷贝问题排查

Blob存储结构与需求说明 #

Blob存储文件夹结构如下：

blob_name
fold1
    2024
        01
        02
        ...
        12
    2023
        01
        02
        ...
        12
fold2
    2024
        01
        02
        ...
        12
    2023
        01
        02
        ...
        12

需求：基于文件最后修改日期筛选"新"文件（示例中为2024-01-01之后的文件），并拷贝到目标存储的对应层级文件夹中。

现有Pipeline问题排查 #

提交的Pipeline JSON #

{"name": "CopyNewFilesPipeline","properties": {"activities": [{"name": "ForEachYear","type": "ForEach","dependsOn": [{"activity": "GetMetadataActivity","dependencyConditions": ["Succeeded"]}],"userProperties": [],"typeProperties": {"items": {"value": "@activity('GetMetadataActivity').output.childItems","type": "Expression"},"activities": [{"name": "ForEachMonth","type": "ForEach","dependsOn": [],"userProperties": [],"typeProperties": {"items": {"value": "@item().childItems","type": "Expression"},"activities": [{"name": "IfCondition","type": "IfCondition","dependsOn": [],"userProperties": [],"typeProperties": {"expression": {"value": "@greaterOrEquals(item().lastModified, '2024-01-01T00:00:00Z')","type": "Expression"},"ifTrueActivities": [{"name": "Copy data1","type": "Copy","dependsOn": [],"policy": {"timeout": "0.12:00:00","retry": 0,"retryIntervalInSeconds": 30,"secureOutput": false,"secureInput": false},"userProperties": [],"typeProperties": {"source": {"type": "JsonSource","storeSettings": {"type": "AzureBlobFSReadSettings","recursive": true,"enablePartitionDiscovery": false},"formatSettings": {"type": "JsonReadSettings"}},"sink": {"type": "JsonSink","storeSettings": {"type": "AzureBlobFSWriteSettings","copyBehavior": "PreserveHierarchy"},"formatSettings": {"type": "JsonWriteSettings"}},"enableStaging": false},"inputs": [{"referenceName": "DS_soruce_JSON1","type": "DatasetReference"}],"outputs": [{"referenceName": "DS_land_JSON1","type": "DatasetReference"}}]}]}}]}},{"name": "GetMetadataActivity","type": "GetMetadata","dependsOn": [],"policy": {"timeout": "0.12:00:00","retry": 0,"retryIntervalInSeconds": 30,"secureOutput": false,"secureInput": false},"userProperties": [],"typeProperties": {"dataset": {"referenceName": "DS_soruce_JSON1","type": "DatasetReference"},"fieldList": ["childItems"],"storeSettings": {"type": "AzureBlobFSReadSettings","enablePartitionDiscovery": false},"formatSettings": {"type": "BinaryReadSettings"}}}]},"annotations": []}

核心问题分析 #

1. 表达式逻辑错误
   • IfCondition中使用item().lastModified，此处的item()指向的是月份文件夹，而非文件夹内的文件，判断的是文件夹的修改时间，不是文件的，完全偏离需求。
2. 数据集未参数化
   • Copy活动的源/目标数据集DS_soruce_JSON1（注意拼写错误：soruce应为source）未动态绑定当前遍历的路径，导致每次执行都会拷贝整个数据源，而非对应年份/月份的文件。
3. BadRequest错误的直接诱因
   • 数据集名称拼写错误（DS_soruce_JSON1）可能导致ADF无法识别数据集；
   • ForEach层级嵌套中，item()的指向混淆，导致表达式返回无效值；
   • GetMetadata仅获取了文件夹的元数据，未递归获取文件信息，后续逻辑无有效数据支撑。

正确流水线构建步骤 #

1. 配置参数化数据集 #

• 源数据集（DS_source_JSON）：添加字符串参数folderPath，路径设置为@concat('blob_name/', dataset().folderPath)，用于接收流水线传递的动态路径。
• 目标数据集（DS_land_JSON）：同样添加字符串参数folderPath，路径设置为@concat('target_blob/', dataset().folderPath)，保持与源路径的层级一致。

2. 流水线结构设计 #

步骤1：获取根目录下的一级文件夹（fold1、fold2）

添加GetMetadata_RootFolders活动：

• 数据集指向Blob根目录（blob_name）；
• 勾选childItems字段，获取根目录下的所有一级文件夹。

步骤2：遍历一级文件夹

添加ForEach_RootFolder活动：

• Items表达式：@activity('GetMetadata_RootFolders').output.childItems；
• 勾选Sequential（可选，如需按顺序遍历）。

步骤3：获取当前一级文件夹下的年份文件夹

在ForEach_RootFolder内部添加GetMetadata_YearFolders活动：

• 源数据集的folderPath参数设置为@item().name；
• 勾选childItems字段，获取年份文件夹列表。

步骤4：遍历年份文件夹

添加ForEach_Year活动：

• Items表达式：@activity('GetMetadata_YearFolders').output.childItems。

步骤5：获取当前年份文件夹下的月份文件夹

在ForEach_Year内部添加GetMetadata_MonthFolders活动：

• 源数据集的folderPath参数设置为@concat(item().parent.name, '/', item().name)；
• 勾选childItems字段，获取月份文件夹列表。

步骤6：遍历月份文件夹并筛选拷贝文件

添加ForEach_Month活动：

• Items表达式：@activity('GetMetadata_MonthFolders').output.childItems；
• 内部添加Copy活动：
  • 源数据集的folderPath参数设置为@concat(item().parent.parent.name, '/', item().parent.name, '/', item().name)；
  • 源存储设置（AzureBlobFSReadSettings）中，添加modifiedDateTimeStart参数，值为'2024-01-01T00:00:00Z'，自动筛选该日期后修改的文件；
  • 目标数据集的folderPath参数与源保持一致，确保层级对应。

3. 优化方案：递归获取所有文件并筛选 #

如果不想多层遍历，可简化流程：

1. 添加GetMetadata_AllFiles活动：
   • 数据集指向根目录，开启recursive: true；
   • 勾选childItems和lastModified字段，获取所有文件的元数据。
2. 添加Filter活动：
   • 筛选条件：@and(greaterOrEquals(item().lastModified, '2024-01-01T00:00:00Z'), equals(item().type, 'File'))，仅保留符合时间条件的文件。
3. 添加ForEach_Files活动：
   • Items表达式：@activity('Filter').output.value；
   • 内部Copy活动的源路径设置为@item().name，目标路径设置为@replace(item().name, 'blob_name/', 'target_blob/')，自动保留层级。

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