使用GCSToGCSOperator在GCS桶移动文件时重复创建目录的问题

核心问题分析 #

你遇到的嵌套文件夹和路径过长问题，主要来自两个原因：

1. GCSToGCSOperator目标路径配置错误：当destination_object不以斜杠/结尾时，Operator会将源文件的完整相对路径拼接到目标路径后，导致嵌套目录；
2. 传感器可能误触发已处理文件：如果传感器没有过滤loaded_files子文件夹内的文件，会重复处理已移动的文件，进一步加剧路径嵌套。

具体修复步骤 #

1. 修正GCSToGCSOperator的目标路径配置 #

将destination_object改为以斜杠结尾的文件夹路径，确保文件直接移动到目标文件夹下，而非拼接嵌套路径：

move_files = GCSToGCSOperator(
    task_id='move_files',
    impersonation_chain='service_account',
    source_bucket='ingestion_bucket',
    source_object='ABCD/Daily_data*.csv',
    destination_bucket='ingestion_bucket',
    destination_object='ABCD/loaded_files/',  # 末尾添加斜杠
    move_object=True
)

这样配置后，匹配到的ABCD/Daily_dataXXX.csv会直接移动到ABCD/loaded_files/Daily_dataXXX.csv，不会生成嵌套的子文件夹。

2. 确保传感器仅监控未处理的文件 #

修改传感器配置，限定只监控ABCD根目录下的Daily_data*.csv，排除loaded_files子文件夹内的文件。以GCSObjectsWithPrefixExistenceSensor为例：

from airflow.providers.google.cloud.sensors.gcs import GCSObjectsWithPrefixExistenceSensor

wait_for_files = GCSObjectsWithPrefixExistenceSensor(
    task_id='wait_for_files',
    bucket='ingestion_bucket',
    prefix='ABCD/Daily_data',  # 仅监控ABCD目录下的目标文件
    exclude_prefixes=['ABCD/loaded_files/'],  # 排除已处理的文件目录
    mode='poke'
)

通过exclude_prefixes参数过滤已处理的文件路径，避免传感器重复触发已移动的文件。

3. 额外优化：避免重复处理的兜底方案 #

如果传感器过滤仍有风险，可依赖move_object=True的特性——该参数会在移动完成后自动删除源文件，只要传感器不监控loaded_files目录，就能从根源避免重复处理。

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