使用transfer_manager上传GCS大文件遇TimeoutError问题求助

问题背景 #

使用Google Cloud Storage官方文档提供的transfer_manager.upload_many_from_filenames方法上传小文件（CSV、XLSX）正常，但上传2000KB左右的CSV文件时触发超时错误：

('Connection aborted.', TimeoutError('The write operation timed out'))

该方法无直接配置上传超时的参数，尝试将bucket = storage_client.bucket(bucket_name)改为bucket = client.get_bucket(BUCKET_NAME, timeout=300.0)，但此改动无法解决上传超时问题（get_bucket的timeout仅作用于桶元数据获取请求，与文件上传无关）。

可行解决方案 #

1. 配置客户端全局超时 #

初始化Client时设置timeout参数，该配置会作用于所有API请求（包括上传操作）：

storage_client = Client(timeout=300.0)  # 设置全局超时为5分钟

2. 启用分块上传 #

大文件单块上传易超时，通过upload_many_from_filenames的chunk_size参数启用分块上传，将文件拆分为多个小块独立上传。注意分块大小需为256KB的整数倍（GCS要求最小分块256KB）：

results = transfer_manager.upload_many_from_filenames(
    bucket, filenames, source_directory=source_directory, 
    max_workers=workers, chunk_size=10*1024*1024  # 10MB分块
)

3. 自定义重试策略 #

针对网络超时等异常，配置自动重试策略，提升上传成功率：

from google.api_core.retry import Retry
from google.cloud.storage.retry import DEFAULT_RETRY

custom_retry = Retry(
    initial=1.0,    # 初始重试间隔（秒）
    total=300.0,    # 总重试时间上限（秒）
    maximum=10.0,   # 最大重试间隔（秒）
    multiplier=2,   # 间隔倍数（指数退避）
    predicate=DEFAULT_RETRY._predicate,  # 沿用默认重试触发条件（含超时）
)

storage_client = Client(retry=custom_retry)

完整修改代码 #

from google.cloud.storage import Client, transfer_manager
from google.api_core.retry import Retry
from google.cloud.storage.retry import DEFAULT_RETRY

def upload_many_blobs_with_transfer_manager(input_logger,
    bucket_name, filenames, source_directory="", workers=8):
    """Upload every file in a list to a bucket, concurrently in a process pool.

    Each blob name is derived from the filename, not including the
    `source_directory` parameter. For complete control of the blob name for each
    file (and other aspects of individual blob metadata), use
    transfer_manager.upload_many() instead.
    """

    # 自定义重试策略，处理超时等网络异常
    custom_retry = Retry(
        initial=1.0,
        total=300.0,
        maximum=10.0,
        multiplier=2,
        predicate=DEFAULT_RETRY._predicate,
    )

    # 初始化客户端，设置全局超时+自定义重试
    storage_client = Client(timeout=300.0, retry=custom_retry)
    bucket = storage_client.bucket(bucket_name)

    results = transfer_manager.upload_many_from_filenames(
        bucket, filenames, source_directory=source_directory, 
        max_workers=workers, chunk_size=10*1024*1024  # 启用10MB分块上传
    )

    for name, result in zip(filenames, results):
        if isinstance(result, Exception):
            input_logger.info("Failed to upload {} due to exception: {}".format(name, result))
        else:
            input_logger.info("Uploaded {} to {}.".format(name, bucket_name))

补充说明 #

• 不要通过get_bucket设置超时，该参数仅影响桶信息查询，与文件上传流程无关。
• 分块上传是解决大文件超时的核心手段，若网络环境较差，可适当调小chunk_size（如5MB）。
• 重试策略可根据实际网络情况调整total和maximum参数，平衡重试次数与等待时间。

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