按国别批量复制文件至双目标端的Pipeline实现及代码建议

核心实现思路 #

通过两层GetMetadata+Foreach循环组合，先遍历源根目录下的所有国别子目录，再遍历每个子目录内的文件，最终用两个Copy Activity分别完成到两个目标存储的同步，全程保持原有的国别目录结构。

Pipeline 配置及代码示例 #

以下是完整的Pipeline JSON定义，包含参数配置、组件逻辑：

{
    "name": "BatchCopy_DualTargets",
    "properties": {
        "activities": [
            {
                "name": "Get Country Folders",
                "type": "GetMetadata",
                "dependsOn": [],
                "policy": {
                    "timeout": "7.00:00:00",
                    "retry": 0,
                    "retryIntervalInSeconds": 30,
                    "secureOutput": false,
                    "secureInput": false
                },
                "userProperties": [],
                "typeProperties": {
                    "dataset": {
                        "referenceName": "Source_FileSystem_Root",
                        "type": "DatasetReference"
                    },
                    "fieldList": [
                        "childItems"
                    ],
                    "storeSettings": {
                        "type": "AzureBlobStorageReadSettings",
                        "recursive": false
                    },
                    "formatSettings": {
                        "type": "DelimitedTextReadSettings"
                    }
                }
            },
            {
                "name": "Loop Through Country Folders",
                "type": "ForEach",
                "dependsOn": [
                    {
                        "activity": "Get Country Folders",
                        "dependencyConditions": [
                            "Succeeded"
                        ]
                    }
                ],
                "userProperties": [],
                "typeProperties": {
                    "items": {
                        "value": "@activity('Get Country Folders').output.childItems",
                        "type": "Expression"
                    },
                    "isSequential": false,
                    "activities": [
                        {
                            "name": "Get Files in Current Folder",
                            "type": "GetMetadata",
                            "dependsOn": [],
                            "policy": {
                                "timeout": "7.00:00:00",
                                "retry": 0,
                                "retryIntervalInSeconds": 30,
                                "secureOutput": false,
                                "secureInput": false
                            },
                            "userProperties": [],
                            "typeProperties": {
                                "dataset": {
                                    "referenceName": "Source_Country_Folder",
                                    "type": "DatasetReference",
                                    "parameters": {
                                        "CountryFolderName": "@item().name"
                                    }
                                },
                                "fieldList": [
                                    "childItems"
                                ],
                                "storeSettings": {
                                    "type": "AzureBlobStorageReadSettings",
                                    "recursive": false
                                },
                                "formatSettings": {
                                    "type": "DelimitedTextReadSettings"
                                }
                            }
                        },
                        {
                            "name": "Loop Through Files",
                            "type": "ForEach",
                            "dependsOn": [
                                {
                                    "activity": "Get Files in Current Folder",
                                    "dependencyConditions": [
                                        "Succeeded"
                                    ]
                                }
                            ],
                            "userProperties": [],
                            "typeProperties": {
                                "items": {
                                    "value": "@activity('Get Files in Current Folder').output.childItems",
                                    "type": "Expression"
                                },
                                "isSequential": false,
                                "activities": [
                                    {
                                        "name": "Copy to Target FileSystem",
                                        "type": "Copy",
                                        "dependsOn": [],
                                        "policy": {
                                            "timeout": "7.00:00:00",
                                            "retry": 2,
                                            "retryIntervalInSeconds": 30,
                                            "secureOutput": false,
                                            "secureInput": false
                                        },
                                        "userProperties": [],
                                        "typeProperties": {
                                            "source": {
                                                "type": "DelimitedTextSource",
                                                "storeSettings": {
                                                    "type": "AzureBlobStorageReadSettings",
                                                    "recursive": false,
                                                    "wildcardFileName": "@item().name",
                                                    "folderPath": {
                                                        "value": "@concat(pipeline().parameters.SourceRootPath, '/', activity('Loop Through Country Folders').items[loopIndex()].name)",
                                                        "type": "Expression"
                                                    }
                                                },
                                                "formatSettings": {
                                                    "type": "DelimitedTextReadSettings"
                                                }
                                            },
                                            "sink": {
                                                "type": "DelimitedTextSink",
                                                "storeSettings": {
                                                    "type": "AzureBlobStorageWriteSettings",
                                                    "folderPath": {
                                                        "value": "@concat(pipeline().parameters.TargetFileSystemRoot, '/', activity('Loop Through Country Folders').items[loopIndex()].name)",
                                                        "type": "Expression"
                                                    }
                                                },
                                                "formatSettings": {
                                                    "type": "DelimitedTextWriteSettings"
                                                }
                                            },
                                            "enableStaging": false,
                                            "translator": {
                                                "type": "TabularTranslator",
                                                "typeConversion": true,
                                                "typeConversionSettings": {
                                                    "allowDataTruncation": true,
                                                    "treatBooleanAsNumber": false
                                                }
                                            }
                                        },
                                        "inputs": [
                                            {
                                                "referenceName": "Source_Country_Folder",
                                                "type": "DatasetReference",
                                                "parameters": {
                                                    "CountryFolderName": "@activity('Loop Through Country Folders').items[loopIndex()].name"
                                                }
                                            }
                                        ],
                                        "outputs": [
                                            {
                                                "referenceName": "Target_FileSystem_Country_Folder",
                                                "type": "DatasetReference",
                                                "parameters": {
                                                    "CountryFolderName": "@activity('Loop Through Country Folders').items[loopIndex()].name"
                                                }
                                            }
                                        ]
                                    },
                                    {
                                        "name": "Copy to ADLS",
                                        "type": "Copy",
                                        "dependsOn": [
                                            {
                                                "activity": "Copy to Target FileSystem",
                                                "dependencyConditions": [
                                                    "Succeeded"
                                                ]
                                            }
                                        ],
                                        "policy": {
                                            "timeout": "7.00:00:00",
                                            "retry": 2,
                                            "retryIntervalInSeconds": 30,
                                            "secureOutput": false,
                                            "secureInput": false
                                        },
                                        "userProperties": [],
                                        "typeProperties": {
                                            "source": {
                                                "type": "DelimitedTextSource",
                                                "storeSettings": {
                                                    "type": "AzureBlobStorageReadSettings",
                                                    "recursive": false,
                                                    "wildcardFileName": "@item().name",
                                                    "folderPath": {
                                                        "value": "@concat(pipeline().parameters.SourceRootPath, '/', activity('Loop Through Country Folders').items[loopIndex()].name)",
                                                        "type": "Expression"
                                                    }
                                                },
                                                "formatSettings": {
                                                    "type": "DelimitedTextReadSettings"
                                                }
                                            },
                                            "sink": {
                                                "type": "DelimitedTextSink",
                                                "storeSettings": {
                                                    "type": "AzureBlobStorageWriteSettings",
                                                    "folderPath": {
                                                        "value": "@concat(pipeline().parameters.ADLSRootPath, '/', activity('Loop Through Country Folders').items[loopIndex()].name)",
                                                        "type": "Expression"
                                                    }
                                                },
                                                "formatSettings": {
                                                    "type": "DelimitedTextWriteSettings"
                                                }
                                            },
                                            "enableStaging": false,
                                            "translator": {
                                                "type": "TabularTranslator",
                                                "typeConversion": true,
                                                "typeConversionSettings": {
                                                    "allowDataTruncation": true,
                                                    "treatBooleanAsNumber": false
                                                }
                                            }
                                        },
                                        "inputs": [
                                            {
                                                "referenceName": "Source_Country_Folder",
                                                "type": "DatasetReference",
                                                "parameters": {
                                                    "CountryFolderName": "@activity('Loop Through Country Folders').items[loopIndex()].name"
                                                }
                                            }
                                        ],
                                        "outputs": [
                                            {
                                                "referenceName": "ADLS_Country_Folder",
                                                "type": "DatasetReference",
                                                "parameters": {
                                                    "CountryFolderName": "@activity('Loop Through Country Folders').items[loopIndex()].name"
                                                }
                                            }
                                        ]
                                    }
                                ]
                            }
                        }
                    ]
                }
            }
        ],
        "parameters": {
            "SourceRootPath": {
                "type": "string",
                "defaultValue": "source-root"
            },
            "TargetFileSystemRoot": {
                "type": "string",
                "defaultValue": "target-filesystem-root"
            },
            "ADLSRootPath": {
                "type": "string",
                "defaultValue": "adls-root"
            }
        },
        "annotations": []
    }
}

关键配置说明 #

• Dataset 配置
  • Source_FileSystem_Root：指向源存储的根目录，用于第一层GetMetadata获取国别子目录
  • Source_Country_Folder：带参数CountryFolderName的数据集，动态指向当前遍历的国别子目录
  • Target_FileSystem_Country_Folder和ADLS_Country_Folder：分别对应两个目标存储的国别子目录数据集，同样带CountryFolderName参数
• 动态路径拼接
  • 源文件路径：通过参数+循环变量拼接，保持源根目录/国别目录/文件名的结构
  • 目标路径：分别用对应目标根目录参数拼接，与源目录结构完全一致
• 并行处理
  • 两层Foreach均设置isSequential: false，可并行处理多个目录和文件，提升同步效率

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