如何使用Airflow S3Hook跨不同连接ID复制S3存储桶对象？

原代码中的copy_object仅支持同一AWS账户/同一连接权限下的对象复制，跨不同连接ID（对应不同AWS账户或权限组）时会因权限校验失败无法执行。以下是两种可行的实现方案：

方案1：临时文件中转（适合小文件） #

通过「下载源对象到本地临时文件→上传至目标桶」的逻辑实现跨连接复制，操作简单直观：

import tempfile
import os
from airflow.providers.amazon.aws.hooks.s3 import S3Hook

# 初始化两个独立的S3连接
target_s3 = S3Hook(self.aws_conn_id)
source_s3 = S3Hook(self.ext_aws_conn_id)

# 创建临时文件存储下载的源对象
with tempfile.NamedTemporaryFile(mode='wb', delete=False) as tmp_file:
    # 从外部S3桶下载对象到临时文件
    source_s3.download_fileobj(
        bucket_name=self.partition.bucket,
        key=key,
        file_obj=tmp_file
    )

# 将临时文件上传到目标S3桶
target_s3.upload_file(
    filename=tmp_file.name,
    bucket_name="bucket",
    key=f'dest_key',
    extra_args={'ContentEncoding': 'csv'}
)

# 清理临时文件
os.unlink(tmp_file.name)

方案2：预签名URL中转（适合大文件） #

利用源连接生成带访问权限的预签名URL，让目标连接通过该URL直接拉取源对象并上传至目标桶，无需本地存储，传输效率更高：

import requests
from airflow.providers.amazon.aws.hooks.s3 import S3Hook

# 初始化两个独立的S3连接
target_s3 = S3Hook(self.aws_conn_id)
source_s3 = S3Hook(self.ext_aws_conn_id)

# 生成源对象的预签名URL（有效期设为3600秒，可按需调整）
presigned_url = source_s3.generate_presigned_url(
    client_method='get_object',
    params={
        'Bucket': self.partition.bucket,
        'Key': key
    },
    expires_in=3600
)

# 通过预签名URL拉取内容并上传到目标桶
with requests.get(presigned_url, stream=True) as r:
    r.raise_for_status()
    target_s3.upload_fileobj(
        file_obj=r.raw,
        bucket_name="bucket",
        key=f'dest_key',
        extra_args={'ContentEncoding': 'csv'}
    )

注意事项 #

• 需确保source_s3连接对源桶拥有读权限，target_s3连接对目标桶拥有写权限
• 方案2的预签名URL有效期需设置足够长，避免上传过程中URL过期失效
• 大文件场景优先选择方案2，避免占用本地磁盘资源

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