如何使用Azure Data Factory从嵌套父文件夹下的子文件夹提取最新日期文件夹

前提说明 #

数据源根路径为 master/item/itemsold/data/V1/，路径下所有子文件夹均以YYYYMMDD格式的日期命名，目标是筛选出修改时间最新的文件夹（示例中为20211201），整体复制到指定的Sink端路径。

不同环境下的实现方法 #

Linux Shell 本地环境实现 #

# 进入数据源根目录
cd master/item/itemsold/data/V1/
# 筛选出修改时间最新的文件夹，先打印确认文件夹名称
latest_dir=$(ls -dt */ | head -n1)
echo $latest_dir
# 确认无误后执行复制，替换为你的实际Sink端路径
cp -r "$latest_dir" /path/to/your/sink/

参数说明：

• ls -dt */：仅列出当前路径下的子文件夹，按修改时间倒序排序
• head -n1：取排序结果的第一条，即最新修改的文件夹
• cp -r：递归复制整个文件夹到目标路径

HDFS 分布式存储环境实现 #

如果数据源和Sink端都在HDFS上，使用Hadoop命令实现：

# 获取HDFS数据源路径下最新修改的文件夹
latest_dir=$(hadoop fs -ls master/item/itemsold/data/V1/ | grep '^d' | sort -k6,7 -r | head -n1 | awk '{print $8}')
# 复制到HDFS Sink路径
hadoop fs -cp -r "$latest_dir" /path/to/hdfs/sink/

Python 自动化脚本实现 #

import os
import shutil

source_root = "master/item/itemsold/data/V1/"
sink_root = "/path/to/your/sink/"

# 筛选根路径下所有子文件夹
dir_list = [d for d in os.listdir(source_root) if os.path.isdir(os.path.join(source_root, d))]
# 按修改时间倒序排序，取最新的文件夹
latest_dir = sorted(dir_list, key=lambda x: os.path.getmtime(os.path.join(source_root, x)), reverse=True)[0]
# 执行复制，Python3.8+可添加dirs_exist_ok=True参数覆盖Sink端已存在的同名文件夹
shutil.copytree(os.path.join(source_root, latest_dir), os.path.join(sink_root, latest_dir))

print(f"已成功将最新文件夹 {latest_dir} 复制到Sink端")

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