如何通过ADF管道将多源HDFS文件复制到ADLS Gen2存储？

需要将HDFS中的ZIP文件批量复制到ADLS Gen2 Blob存储，路径转换规则如下：

HDFS源路径示例：

hdfs/data/users/synova/raw/partition1/customer/full/2023-04-01/customer.zip
hdfs/data/users/synova/raw/partition1/customer/full/2023-04-02/customer.zip
hdfs/data/users/synova/raw/partition2/parts/full/2023-04-01/parts.zip

ADLS目标路径示例：

adls/consolidated/synova/raw/partition1/customer/2023-04-01/customer.zip
adls/consolidated/synova/raw/partition1/customer/2023-04-02/customer.zip
adls/consolidated/synova/raw/partition2/parts/2023-04-01/parts.zip

核心转换规则：

1. 根前缀从hdfs/data/users/替换为adls/consolidated/
2. 移除路径中的/full/层级

通过ADF的内置表达式语言实现动态路径转换，无需硬编码路径片段，确保管道的通用性：

• 先替换根前缀，再移除/full/片段
• 或通过拆分路径数组、过滤掉full元素后重组路径，适配更灵活的结构调整

1. 配置链接服务 #

• HDFS链接服务：创建指向目标Hadoop集群的链接服务，确认ADF具备HDFS资源的访问权限。
• ADLS Gen2链接服务：创建基于账户密钥或托管身份验证的ADLS Gen2链接服务，关联目标存储账户。

2. 添加Get Metadata活动 #

• 命名为Get_HDFS_Zip_Files
• 数据源选择HDFS链接服务，数据集指向HDFS根目录hdfs/data/users/synova/raw/，勾选递归选项以遍历所有子目录
• 字段列表中选择Child Items，用于获取所有ZIP文件的完整路径信息

3. 添加For Each活动 #

• 将For Each的输入设置为@activity('Get_HDFS_Zip_Files').output.childItems
• 可根据需求勾选Sequential（顺序处理）或保持并行处理模式

4. 在For Each内部添加Copy活动 #

4.1 配置Copy活动源 #

• 数据源选择HDFS链接服务
• 数据集设置为“文件路径”模式，文件路径使用表达式：@item().path（指向当前遍历到的HDFS文件）

4.2 配置Copy活动目标 #

• 目标选择ADLS Gen2链接服务
• 数据集设置为“文件路径”模式，文件路径使用以下表达式完成转换：

@replace(replace(item().path, 'hdfs/data/users/', 'adls/consolidated/'), '/full/', '/')

• 若需适配更灵活的路径结构，可使用拆分重组方式：

@join(
    filter(
        split(item().path, '/'), 
        not(equals(item, 'full'))
    ), 
    '/'
)

5. 测试与验证 #

触发管道运行，检查ADLS Gen2目标路径下的文件是否按预期结构存储，确认所有ZIP文件复制完成且路径符合规则。

• 失败重试：在Copy活动设置中配置重试次数，应对临时网络或访问问题
• 日志记录：在For Each中添加Set Variable活动，记录每个文件的复制状态，便于排查问题
• 增量复制：若仅需复制新增文件，可在Get Metadata后添加Filter活动，基于文件修改时间筛选目标文件

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