大文件从S3到Google Cloud Storage的内存高效传输代码问询

嘿，我完全懂你的需求——绕开内存中转，直接让S3和GCS之间流式传文件，毕竟把大文件塞进内存既费资源又容易崩。而且你明确说了不想用Google的传输服务或CLI，要自己写代码实现，刚好我有几个靠谱的方案给你：

关键就是直接获取S3对象的字节流，把这个流作为GCS上传的数据源，让数据一块一块从S3拉取，直接推送到GCS，全程不会把整个文件加载到内存里。

方案1：利用SDK原生流式能力（推荐，以Python为例） #

用boto3（AWS SDK）和google-cloud-storage的原生API就能实现，代码简洁还靠谱，完全贴合你拆分小方法的需求：

拆分后的小方法示例 #

import boto3
from google.cloud import storage
from botocore.config import Config

def get_s3_object_stream(s3_bucket: str, s3_key: str) -> any:
    """获取S3对象的流式响应，不加载到内存"""
    # 添加重试配置提升稳定性
    s3_config = Config(retries={'max_attempts': 3, 'mode': 'standard'})
    s3_client = boto3.client('s3', config=s3_config)
    # 返回的Body是可迭代的字节流对象
    return s3_client.get_object(Bucket=s3_bucket, Key=s3_key)['Body']

def upload_stream_to_gcs(gcs_bucket: str, gcs_blob_name: str, data_stream: any):
    """将流式数据上传到GCS，支持大文件自动分段"""
    gcs_client = storage.Client()
    bucket = gcs_client.bucket(gcs_bucket)
    blob = bucket.blob(gcs_blob_name)
    
    # 10MB chunk size，可根据网络情况调整
    blob.upload_from_file(data_stream, chunk_size=10*1024*1024)

def transfer_s3_to_gcs(s3_bucket: str, s3_key: str, gcs_bucket: str, gcs_blob_name: str):
    """主方法：串联S3流获取和GCS流上传"""
    s3_stream = get_s3_object_stream(s3_bucket, s3_key)
    upload_stream_to_gcs(gcs_bucket, gcs_blob_name, s3_stream)

为什么这个方案高效？ #

• s3_client.get_object()['Body'] 返回的不是完整文件，而是一个可迭代的字节流，每次只会拉取一小部分数据
• GCS的upload_from_file() 接受任何类文件对象（实现了read()方法），会自动把流里的数据分块上传，不会一次性加载到内存
• 你可以轻松调整chunk_size来平衡网络效率和内存占用

方案2：手动处理分段上传（适合超大型文件） #

如果你的文件是几十GB甚至上百GB的超大文件，可以手动控制分段上传逻辑，更灵活：

def transfer_large_file_s3_to_gcs(s3_bucket: str, s3_key: str, gcs_bucket: str, gcs_blob_name: str, chunk_size: int = 32*1024*1024):
    s3_client = boto3.client('s3')
    gcs_client = storage.Client()
    bucket = gcs_client.bucket(gcs_bucket)
    blob = bucket.blob(gcs_blob_name)
    
    # 初始化GCS分段上传
    upload_session = blob.initiate_resumable_upload()
    
    # 分段读取S3数据并上传
    while True:
        start_byte = upload_session.total_bytes_uploaded
        end_byte = start_byte + chunk_size - 1
        chunk = s3_client.get_object(Bucket=s3_bucket, Key=s3_key, Range=f'bytes={start_byte}-{end_byte}')['Body'].read()
        if not chunk:
            break
        upload_session.write(chunk)
    
    # 完成上传
    upload_session.finish()

这个方法适合需要精确控制分段大小、断点续传的场景，不过一般来说方案1的自动分段已经足够用了。

注意事项 #

• 权限配置：确保运行代码的身份同时拥有S3的s3:GetObject权限和GCS的storage.objects.create权限
• 网络优化：如果你的运行环境在某个云厂商内部，尽量用内网端点访问（比如AWS VPC内访问S3，GCP VPC内访问GCS），能大幅提升速度
• 错误处理：可以给两个SDK添加重试机制，或者捕获异常后重试上传，避免网络波动导致失败

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