如何使用GCSToS3Operator按文件名从GCS传输单个文件到S3

直接通过source_object参数指定单个文件名即可，无需依赖prefix或delimiter（这两个参数用于批量匹配文件），具体实现如下：

核心实现代码 #

from airflow import DAG
from airflow.providers.google.cloud.transfers.gcs_to_s3 import GCSToS3Operator
from datetime import datetime

with DAG(
    dag_id="gcs_single_file_to_s3",
    start_date=datetime(2024, 1, 1),
    schedule_interval=None,
    catchup=False
) as dag:
    transfer_task = GCSToS3Operator(
        task_id="transfer_single_csv",
        # GCS配置
        gcs_bucket="your-gcs-bucket-name",
        source_object="your-target-file.csv",  # 直接指定GCS桶内的单个文件名
        # S3配置
        dest_bucket="your-s3-bucket-name",
        dest_object="your-target-file.csv",  # S3端的目标文件名，可自定义
        # 连接ID（需提前在Airflow中配置）
        gcp_conn_id="google_cloud_default",
        aws_conn_id="aws_default"
    )

参数说明 #

• source_object：直接填写GCS桶内的目标文件名（因你的桶无文件夹结构，无需加路径前缀），Operator会仅传输该文件
• dest_object：可选参数，若省略则默认使用source_object的文件名作为S3端的文件名
• 无需设置prefix和delimiter，这两个参数是用于批量匹配一类文件（比如所有CSV或某个路径下的文件），不符合单个文件传输的需求

注意事项 #

• 确保Airflow已正确配置GCP和AWS的连接（对应gcp_conn_id和aws_conn_id）
• 若目标文件在S3端已存在，默认会覆盖，如需避免覆盖可添加replace=False参数

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