如何使用Data Factory按YYYY-MM格式创建月度文件夹并复制数据？

核心需求梳理 #

• 源：数据湖内某文件夹，每月新增多组月度数据文件
• 目标：在数据湖另一文件夹下，按YYYY-MM格式自动创建子文件夹，将对应月份的文件复制归档
• 执行频率：每月运行一次，跟随月度数据加载节奏

实现方案（分工具场景） #

场景1：用云数据工厂实现（以Azure Data Factory为例） #

1. 配置参数化数据集

• 源数据集：指向源文件夹，设置文件路径为通配符（比如source/monthly_files/*），支持匹配所有当月新增文件
• 目标数据集：路径设置为参数化格式 target/archived_data/@formatDateTime(utcnow(), 'yyyy-MM')，自动生成当前年月的归档文件夹

2. 构建复制活动

• 新建管道，添加复制活动，源选配置好的源数据集，目标选参数化的目标数据集
• 若源文件名包含月份标识（如sales_2024-05.parquet），可在源数据集的筛选条件中添加contains(name, '@formatDateTime(utcnow(), ''yyyy-MM'')')，精准复制当月文件

3. 设置月度调度

• 创建时间触发器，设置为每月固定日期执行（比如每月5日，确保当月数据加载完成），触发时间根据你的数据加载周期调整

场景2：用Python脚本实现（适配ADLS Gen2、S3等主流数据湖） #

核心脚本示例（以ADLS Gen2为例）

import datetime
from azure.storage.filedatalake import DataLakeServiceClient

# 替换为你的数据湖信息
STORAGE_ACCOUNT = "your_storage_account"
ACCOUNT_KEY = "your_account_key"
SOURCE_CONTAINER = "source_container"
SOURCE_FOLDER = "monthly_raw_data"
TARGET_CONTAINER = "target_container"

# 获取当前年月（如果需要按文件创建时间取月份，可替换为文件元数据的时间）
current_month = datetime.datetime.now().strftime("%Y-%m")
target_folder = f"archived_by_month/{current_month}"

# 初始化客户端
service_client = DataLakeServiceClient(
    account_url=f"https://{STORAGE_ACCOUNT}.dfs.core.windows.net",
    credential=ACCOUNT_KEY
)

# 获取源文件夹下的所有文件
source_dir_client = service_client.get_file_system_client(SOURCE_CONTAINER).get_directory_client(SOURCE_FOLDER)
files = [f for f in source_dir_client.get_files() if not f.is_directory]

# 创建目标文件夹（不存在则创建）
target_dir_client = service_client.get_file_system_client(TARGET_CONTAINER).get_directory_client(target_folder)
target_dir_client.create_directory()

# 批量复制文件
for file in files:
    # 可选：筛选当月创建的文件，避免复制历史文件
    # file_creation_time = file.properties.created_on.strftime("%Y-%m")
    # if file_creation_time != current_month:
    #     continue
    
    source_file_client = source_dir_client.get_file_client(file.name)
    target_file_client = target_dir_client.get_file_client(file.name)
    
    # 流式复制（大文件推荐用这种方式，避免内存占用过高）
    with source_file_client.download_file() as source_file:
        target_file_client.upload_data(source_file, overwrite=True)

print(f"✅ 当月文件已全部复制到 {target_folder}")

调度设置

• Linux系统：用Cron表达式设置每月执行，比如0 2 5 * * /usr/bin/python3 /path/to/your/script.py（每月5日凌晨2点执行）
• Windows系统：用「任务计划程序」创建基本任务，设置触发频率为每月，指定脚本路径

关键注意事项 #

• 避免重复复制：如果源文件夹不会清理历史文件，一定要添加筛选逻辑（按文件名月份、文件创建时间），防止每次执行都复制所有历史文件
• 权限配置：确保执行管道/脚本的账号拥有数据湖源文件夹的读权限、目标文件夹的写权限
• 异常处理：生产环境建议添加告警机制（比如邮件、企业微信通知），复制失败时及时触发通知

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