如何在Azure Data Factory中按参数指定的文件夹名排除复制文件夹及文件

方案1：Python脚本实现（通用灵活） #

直接写个脚本接收排除参数，遍历源目录时跳过指定邦的文件夹，复制剩余内容。

import os
import shutil
import argparse

def copy_india_states(source_dir, dest_dir, exclude_states):
    if not os.path.isdir(source_dir):
        raise ValueError(f"源目录 {source_dir} 不存在")
    
    os.makedirs(dest_dir, exist_ok=True)
    
    for state_folder in os.listdir(source_dir):
        state_path = os.path.join(source_dir, state_folder)
        if os.path.isdir(state_path) and state_folder not in exclude_states:
            dest_state_path = os.path.join(dest_dir, state_folder)
            shutil.copytree(state_path, dest_state_path, dirs_exist_ok=True)
            print(f"已复制: {state_folder}")

if __name__ == "__main__":
    parser = argparse.ArgumentParser(description='复制印度各邦数据，排除指定邦')
    parser.add_argument('--source', default='./India', help='源India文件夹路径')
    parser.add_argument('--dest', required=True, help='目标存储路径')
    parser.add_argument('--exclude', nargs='+', required=True, help='要排除的邦名列表，多个用空格分隔')
    
    args = parser.parse_args()
    copy_india_states(args.source, args.dest, args.exclude)

用法：
在终端运行脚本，传入目标路径和要排除的邦名：

python copy_script.py --dest ./India_Backup --exclude Kerala Karnataka Odisha

方案2：rsync命令行工具（适合Linux/macOS） #

用系统自带的rsync工具，原生支持排除规则，无需写脚本，效率更高。

单排除项命令：

rsync -av --exclude='Kerala' ./India/ ./India_Backup/

多排除项命令：

rsync -av --exclude='Kerala' --exclude='Karnataka' --exclude='Odisha' ./India/ ./India_Backup/

批量处理排除列表：
如果排除项较多，可以用变量批量生成参数：

EXCLUDE_STATES=("Kerala" "Karnataka" "Odisha")
RSYNC_EXCLUDE_ARGS=()
for state in "${EXCLUDE_STATES[@]}"; do
    RSYNC_EXCLUDE_ARGS+=("--exclude=$state")
done
rsync -av "${RSYNC_EXCLUDE_ARGS[@]}" ./India/ ./India_Backup/

说明：-a是归档模式（保留权限、修改时间等元数据），-v显示复制进度；注意源路径末尾的/表示复制文件夹内的内容，不带/会把整个India文件夹复制到目标目录下。

方案3：ETL工具集成（以Apache Airflow为例） #

如果是在Airflow这类调度工具里构建管道，可以用PythonOperator调用自定义函数，或者用BashOperator执行rsync命令。

PythonOperator示例：

from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime
import os
import shutil

def copy_india_states(**context):
    source_dir = context['params']['source_dir']
    dest_dir = context['params']['dest_dir']
    exclude_states = context['params']['exclude_states']
    
    os.makedirs(dest_dir, exist_ok=True)
    for state_folder in os.listdir(source_dir):
        state_path = os.path.join(source_dir, state_folder)
        if os.path.isdir(state_path) and state_folder not in exclude_states:
            dest_state_path = os.path.join(dest_dir, state_folder)
            shutil.copytree(state_path, dest_state_path, dirs_exist_ok=True)

with DAG(
    'india_state_data_copy',
    start_date=datetime(2024,1,1),
    schedule_interval=None,
    params={
        'source_dir': './India',
        'dest_dir': './India_Backup',
        'exclude_states': ['Kerala', 'Karnataka', 'Odisha']
    }
) as dag:
    copy_task = PythonOperator(
        task_id='copy_exclude_states',
        python_callable=copy_india_states,
        provide_context=True
    )

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