为FTP到Blob的Copy Activity添加序号并解决序号未递增问题

需求说明 #

需要将FTP服务器中的文件复制到Azure Blob存储，同时生成包含以下字段的执行状态记录表：

• S.No：序号
• File Name：文件名
• Error Message：错误信息（无错误则为NA）
• Copy Status：复制状态（成功/失败）

预期输出表格格式：

问题描述 #

编写的Azure Data Factory管道配置中，ForEach循环内的序号（S.No）未按预期递增，多个文件的序号重复。原管道配置如下：

{"name": "ABC_copy2","properties": {"activities": [{"name": "Get Metadata","type": "GetMetadata","dependsOn": [{"activity": "Initialize counter to 1 for SNo 2x","dependencyConditions": ["Succeeded"]}],"policy": {"timeout": "7.00:00:00","retry": 0,"retryIntervalInSeconds": 30,"secureOutput": false,"secureInput": false},"userProperties": [],"typeProperties": {"dataset": {"referenceName": "PH_FTP2_Dir","type": "DatasetReference","parameters": {"foldername": "ABC"}},"fieldList": ["childItems"],"storeSettings": {"type": "FtpReadSettings","recursive": true,"enablePartitionDiscovery": false,"useBinaryTransfer": true,"disableChunking": false},"formatSettings": {"type": "DelimitedTextReadSettings"}}},{"name": "ForEach_ABC","type": "ForEach","dependsOn": [{"activity": "Get Metadata","dependencyConditions": ["Succeeded"]}],"userProperties": [],"typeProperties": {"items": {"value": "@activity('Get Metadata').output.childItems","type": "Expression"},"activities": [{"name": "Move to Blobs","type": "Copy","dependsOn": [],"policy": {"timeout": "7.00:00:00","retry": 0,"retryIntervalInSeconds": 30,"secureOutput": false,"secureInput": false},"userProperties": [],"typeProperties": {"source": {"type": "DelimitedTextSource","storeSettings": {"type": "FtpReadSettings","recursive": true,"enablePartitionDiscovery": false,"useBinaryTransfer": true,"disableChunking": true},"formatSettings": {"type": "DelimitedTextReadSettings"}},"sink": {"type": "DelimitedTextSink","storeSettings": {"type": "AzureBlobStorageWriteSettings"},"formatSettings": {"type": "DelimitedTextWriteSettings","quoteAllText": true,"fileExtension": ".txt"}},"enableStaging": false,"enableSkipIncompatibleRow": true,"logSettings": {"enableCopyActivityLog": true,"copyActivityLogSettings": {"logLevel": "Info","enableReliableLogging": false},"logLocationSettings": {"linkedServiceName": {"referenceName": "AzureBlobStorage_dir","type": "LinkedServiceReference"},"path": "dir"}},"translator": {"type": "TabularTranslator","typeConversion": true,"typeConversionSettings": {"allowDataTruncation": false,"treatBooleanAsNumber": false,"dateTimeFormat": "yyyy-MM-dd HH:mm:ss.SSS"}}},"inputs": [{"referenceName": "PH_FTP2_CopyActivity","type": "DatasetReference","parameters": {"Filename": {"value": "@item().name","type": "Expression"},"Foldername": "ABC"}}],"outputs": [{"referenceName": "PH_AzureStorage","type": "DatasetReference","parameters": {"filename": {"value": "@item().name","type": "Expression"},"foldername": "inputfiles-ABC"}}]},{"name": "Cpy blob Activity Success Status","type": "AppendVariable","dependsOn": [{"activity": "Move to Blobs","dependencyConditions": ["Succeeded"]}],"userProperties": [],"typeProperties": {"variableName": "cpy_blob_status","value": {"value": "@json(concat('{','\"S.No\":',variables('counter'),',\"File Name\":\"',item().Name,'\",\"Copy Status to Blobs\":\"Copied Successfully\"}'))","type": "Expression"}}},{"name": "Cpy blob Activity Failure Status","type": "AppendVariable","dependsOn": [{"activity": "Move to Blobs","dependencyConditions": ["Failed"]}],"userProperties": [],"typeProperties": {"variableName": "cpy_blob_status","value": {"value": "@json(concat('{','\"S.No\":',variables('counter'),',\"File Name\":\"',item().Name,'\",\"Copy Status to Blobs\":\"Failed\",\"Error\":\"',activity('Move to Blobs').output.errors[0].Message,'\"}'))\n","type": "Expression"}}},{"name": "increment and store in temp 2x","type": "SetVariable","dependsOn": [{"activity": "Cpy blob Activity Success Status","dependencyConditions": ["Succeeded"]}],"policy": {"secureOutput": false,"secureInput": false},"userProperties": [],"typeProperties": {"variableName": "temp","value": {"value": "@string(add(int(variables('counter')),1))","type": "Expression"}}},{"name": "store temp to counter 2x","type": "SetVariable","dependsOn": [{"activity": "increment and store in temp 2x","dependencyConditions": ["Succeeded"]}],"policy": {"secureOutput": false,"secureInput": false},"userProperties": [],"typeProperties": {"variableName": "counter","value": {"value": "@variables('temp')","type": "Expression"}}},{"name": "increment and store in temp 2x_F","type": "SetVariable","dependsOn": [{"activity": "Cpy blob Activity Failure Status","dependencyConditions": ["Succeeded"]}],"policy": {"secureOutput": false,"secureInput": false},"userProperties": [],"typeProperties": {"variableName": "temp","value": {"value": "@string(add(int(variables('counter')),1))","type": "Expression"}}},{"name": "store temp to counter 2x_F","type": "SetVariable","dependsOn": [{"activity": "increment and store in temp 2x_F","dependencyConditions": ["Succeeded"]}],"policy": {"secureOutput": false,"secureInput": false},"userProperties": [],"typeProperties": {"variableName": "counter","value": {"value": "@variables('temp')","type": "Expression"}}}]}},{"name": "Initialize counter to 1 for SNo 2x","type": "SetVariable","dependsOn": [],"policy": {"secureOutput": false,"secureInput": false},"userProperties": [],"typeProperties": {"variableName": "counter","value": "1"}}},{"name": "Set Cpy_Blob_Status_final","type": "SetVariable","dependsOn": [{"activity": "ForEach_ABC","dependencyConditions": ["Succeeded"]}],"policy": {"secureOutput": false,"secureInput": false},"userProperties": [],"typeProperties": {"variableName": "cpy_blob_status_final","value": {"value": "@variables('cpy_blob_status')","type": "Expression"}}}]},"variables": {"counter": {"type": "String"},"temp": {"type": "String"},"cpy_blob_status": {"type": "Array"},"cpy_blob_status_final": {"type": "Array"}},"annotations": [],"lastPublishTime": "2023-08-17T08:35:26Z"}},"type": "Microsoft.DataFactory/factories/pipelines"}

问题原因 #

ADF的ForEach活动默认是并行执行（并行度为20），多个迭代会同时读取和修改counter变量，导致变量值被覆盖，无法实现正确的递增逻辑。

解决方案 #

1. 开启ForEach串行执行 #

修改ForEach活动的isSequential属性为true，确保迭代逐个执行，避免变量冲突。

2. 简化递增逻辑 #

去掉冗余的temp变量，直接在SetVariable中更新counter，减少不必要的步骤。

3. 统一成功/失败后的递增流程 #

将递增操作合并为一个活动，通过依赖条件关联成功/失败分支，避免重复代码。

修改后的管道配置 #

{"name": "ABC_copy2","properties": {"activities": [{"name": "Get Metadata","type": "GetMetadata","dependsOn": [{"activity": "Initialize counter to 1 for SNo 2x","dependencyConditions": ["Succeeded"]}],"policy": {"timeout": "7.00:00:00","retry": 0,"retryIntervalInSeconds": 30,"secureOutput": false,"secureInput": false},"userProperties": [],"typeProperties": {"dataset": {"referenceName": "PH_FTP2_Dir","type": "DatasetReference","parameters": {"foldername": "ABC"}},"fieldList": ["childItems"],"storeSettings": {"type": "FtpReadSettings","recursive": true,"enablePartitionDiscovery": false,"useBinaryTransfer": true,"disableChunking": false},"formatSettings": {"type": "DelimitedTextReadSettings"}}},{"name": "ForEach_ABC","type": "ForEach","dependsOn": [{"activity": "Get Metadata","dependencyConditions": ["Succeeded"]}],"userProperties": [],"typeProperties": {"items": {"value": "@activity('Get Metadata').output.childItems","type": "Expression"},"isSequential": true,
"activities": [{"name": "Move to Blobs","type": "Copy","dependsOn": [],"policy": {"timeout": "7.00:00:00","retry": 0,"retryIntervalInSeconds": 30,"secureOutput": false,"secureInput": false},"userProperties": [],"typeProperties": {"source": {"type": "DelimitedTextSource","storeSettings": {"type": "FtpReadSettings","recursive": true,"enablePartitionDiscovery": false,"useBinaryTransfer": true,"disableChunking": true},"formatSettings": {"type": "DelimitedTextReadSettings"}},"sink": {"type": "DelimitedTextSink","storeSettings": {"type": "AzureBlobStorageWriteSettings"},"formatSettings": {"type": "DelimitedTextWriteSettings","quoteAllText": true,"fileExtension": ".txt"}},"enableStaging": false,"enableSkipIncompatibleRow": true,"logSettings": {"enableCopyActivityLog": true,"copyActivityLogSettings": {"logLevel": "Info","enableReliableLogging": false},"logLocationSettings": {"linkedServiceName": {"referenceName": "AzureBlobStorage_dir","type": "LinkedServiceReference"},"path": "dir"}},"translator": {"type": "TabularTranslator","typeConversion": true,"typeConversionSettings": {"allowDataTruncation": false,"treatBooleanAsNumber": false,"dateTimeFormat": "yyyy-MM-dd HH:mm:ss.SSS"}}},"inputs": [{"referenceName": "PH_FTP2_CopyActivity","type": "DatasetReference","parameters": {"Filename": {"value": "@item().name","type": "Expression"},"Foldername": "ABC"}}],"outputs": [{"referenceName": "PH_AzureStorage","type": "DatasetReference","parameters": {"filename": {"value": "@item().name","type": "Expression"},"foldername": "inputfiles-ABC"}}]},{"name": "Cpy blob Activity Success Status","type": "AppendVariable","dependsOn": [{"activity": "Move to Blobs","dependencyConditions": ["Succeeded"]}],"userProperties": [],"typeProperties": {"variableName": "cpy_blob_status","value": {"value": "@json(concat('{\"S.No\":', variables('counter'), ',\"File Name\":\"', item().name, '\",\"Error Message\":\"NA\",\"Copy Status\":\"Successful\"}'))","type": "Expression"}}},{"name": "Cpy blob Activity Failure Status","type": "AppendVariable","dependsOn": [{"activity": "Move to Blobs","dependencyConditions": ["Failed"]}],"userProperties": [],"typeProperties": {"variableName": "cpy_blob_status","value": {"value": "@json(concat('{\"S.No\":', variables('counter'), ',\"File Name\":\"', item().name, '\",\"Error Message\":\"', replace(activity('Move to Blobs').output.errors[0].Message, '\"', '\\\"'), '\",\"Copy Status\":\"Failed\"}'))","type": "Expression"}}},{"name": "Increment Counter","type": "SetVariable","dependsOn": [{"activity": "Cpy blob Activity Success Status","dependencyConditions": ["Succeeded"]}, {"activity": "Cpy blob Activity Failure Status","dependencyConditions": ["Succeeded"]}],"policy": {"secureOutput": false,"secureInput": false},"userProperties": [],"typeProperties": {"variableName": "counter","value": {"value": "@string(add(int(variables('counter')), 1))","type": "Expression"}}}]}},{"name": "Initialize counter to 1 for SNo 2x","type": "SetVariable","dependsOn": [],"policy": {"secureOutput": false,"secureInput": false},"userProperties": [],"typeProperties": {"variableName": "counter","value": "1"}}},{"name": "Set Cpy_Blob_Status_final","type": "SetVariable","dependsOn": [{"activity": "ForEach_ABC","dependencyConditions": ["Succeeded"]}],"policy": {"secureOutput": false,"secureInput": false},"userProperties": [],"typeProperties": {"variableName": "cpy_blob_status_final","value": {"value": "@variables('cpy_blob_status')","type": "Expression"}}}]},"variables": {"counter": {"type": "String"},"cpy_blob_status": {"type": "Array"},"cpy_blob_status_final": {"type": "Array"}},"annotations": [],"lastPublishTime": "2023-08-17T08:35:26Z"}},"type": "Microsoft.DataFactory/factories/pipelines"}

关键修改点说明 #

• ForEach活动添加"isSequential": true，强制串行执行
• 移除temp变量，直接在Increment Counter活动中更新counter
• 统一成功/失败后的递增逻辑，通过多依赖条件触发同一个递增活动
• 调整状态记录的JSON字段，与预期表格的字段名完全匹配（Error Message、Copy Status）
• 对错误信息中的双引号进行转义处理，避免JSON格式错误

验证效果 #

修改后执行管道，序号会按1、2、3依次递增，生成的状态记录与预期表格格式一致，每个文件的状态都能正确标记。

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