Azure Data Factory子目录元数据动态获取及最新文件复制问题

问题1：动态参数化目录名称（固定SFTP_FOLDER/DIRECTORY-B路径） #

按以下步骤配置即可实现：

• 第一步：获取DIRECTORY-B的子目录列表
  使用Get Metadata活动，指向固定路径SFTP_FOLDER/DIRECTORY-B，勾选Child items和Item type属性。此步骤会返回该目录下所有子项的类型和名称。
• 第二步：过滤并遍历子目录
  在ForEach活动中，设置迭代器为过滤后的子目录集合（以ADF为例）：

  @filter(activity('Get Metadata').output.childItems, equals(item().type, 'Folder'))
  

  确保只遍历DIRECTORY-B下的子目录，排除无关项。
• 第三步：动态拼接子目录路径
  在ForEach内部的Get File List活动中，将路径参数设置为动态拼接表达式：

  @concat('SFTP_FOLDER/DIRECTORY-B/', item().name)
  

  验证表达式语法无误且SFTP账号拥有子目录访问权限，即可成功检索对应子目录下的文件。

问题2：递归获取每个子目录最新文件并复制的简便方法 #

方法1：Python脚本实现（灵活可控） #

借助paramiko库连接SFTP，递归遍历目录并筛选每个子目录的最新文件，示例代码如下：

import paramiko
from stat import S_ISDIR

# 获取指定目录下最新修改的文件
def get_latest_file(sftp, remote_dir):
    file_info = []
    for entry in sftp.listdir_attr(remote_dir):
        if not S_ISDIR(entry.st_mode):
            file_info.append((entry.filename, entry.st_mtime))
    if not file_info:
        return None
    # 按修改时间降序排序，取最新文件
    file_info.sort(key=lambda x: x[1], reverse=True)
    return file_info[0][0]

# 递归遍历目录并复制最新文件
def traverse_and_copy(sftp, source_root, target_root):
    for entry in sftp.listdir_attr(source_root):
        source_path = f"{source_root}/{entry.filename}"
        if S_ISDIR(entry.st_mode):
            # 递归进入子目录
            traverse_and_copy(sftp, source_path, f"{target_root}/{entry.filename}")
            latest_file = get_latest_file(sftp, source_path)
            if latest_file:
                # 确保目标目录存在
                try:
                    sftp.mkdir(f"{target_root}/{entry.filename}")
                except IOError:
                    pass  # 目录已存在则跳过
                # 复制文件到目标路径
                sftp.get(f"{source_path}/{latest_file}", f"{target_root}/{entry.filename}/{latest_file}")

# SFTP连接配置
transport = paramiko.Transport(('sftp_host', 22))
transport.connect(username='your_username', password='your_password')
sftp = paramiko.SFTPClient.from_transport(transport)

# 执行复制逻辑
source_dir = "/SFTP_FOLDER/DIRECTORY-B"
target_dir = "/your_target_directory"
traverse_and_copy(sftp, source_dir, target_dir)

# 关闭连接
sftp.close()
transport.close()

方法2：低代码ETL工具实现（以Azure Data Factory为例） #

• 步骤1：递归获取所有子项
  使用Get Metadata活动，指向SFTP_FOLDER/DIRECTORY-B，勾选Recursive、Child items、Item type和Last modified属性。
• 步骤2：筛选文件并分组
  用Filter活动筛选出Item type为File的子项；再用Aggregate活动按文件所在目录分组，保留组内所有文件信息。
• 步骤3：筛选每组最新文件
  对每个分组的文件，用Sort活动按Last modified降序排序，取第一个条目即为该目录下的最新文件。
• 步骤4：复制文件
  使用Copy活动，动态拼接源文件路径和目标路径，完成文件复制。

内容的提问来源于stack exchange，提问作者Mr. Hobo