Airflow使用GCS到GCS算子归档文件产生递归文件夹如何解决

原来的配置中source_object='*.json'没有限制匹配范围，operator会递归扫描整个存储桶下所有后缀为.json的对象，包含Archive子目录下已经归档的文件。每次任务运行时都会把Archive目录内的json文件再次往Archive/路径下移动，最终形成无限嵌套的目录结构。

有两种修改方式均可实现仅移动根目录下的json文件：

方案1：指定根目录匹配规则 #

修改source_object参数为/*.json，前缀斜杠表示仅匹配存储桶根目录下的json文件，不会命中子目录内的文件。
修改后的代码如下：

archive_files = GoogleCloudStorageToGoogleCloudStorageOperator(
    task_id='Archive_Files',
    source_bucket='my_data',
    source_object='/*.json',
    destination_bucket='my_data',
    destination_object='Archive/',
    move_object=True,
    google_cloud_storage_conn_id='connection1',
    dag=dag
)

方案2：添加目录分隔符限制递归 #

新增delimiter='/'参数，告知operator不要递归扫描子目录，仅在当前指定的根目录层级匹配对象。
修改后的代码如下：

archive_files = GoogleCloudStorageToGoogleCloudStorageOperator(
    task_id='Archive_Files',
    source_bucket='my_data',
    source_object='*.json',
    destination_bucket='my_data',
    destination_object='Archive/',
    move_object=True,
    delimiter='/',
    google_cloud_storage_conn_id='connection1',
    dag=dag
)

如果你使用的是Airflow 2.x及以上版本，GoogleCloudStorageToGoogleCloudStorageOperator已更名为GCSToGCSOperator，上述参数配置逻辑完全兼容，直接替换类名即可正常使用。

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