Python实现GCS大压缩文件低内存流式解压分块上传

云原生阿岚

2026-8-28

问题根因

直接对GCS Blob对象调用shutil.copyfileobj失败、内存超标的核心原因有三个：

GCS Python SDK返回的原生Blob对象没有实现标准IO库要求的read(size)/write(size)流式接口，直接传入会触发SDK默认的全量内容缓存逻辑
标准库解压模块（tarfile/zipfile）默认使用随机访问模式读取压缩包索引，非流式打开时会尝试把全量文件加载到内存做seek操作
未显式指定SDK分块大小、流模式参数时，GCS上传下载逻辑默认会缓存全量对象内容，无法实现低内存流式处理

低内存流式实现方案

全程内存占用可稳定控制在32MB以内（可自行调整块大小参数），不需要将文件落地到本地磁盘，适配2GB内存上限的运行环境：

首先安装依赖：

pip install google-cloud-storage

核心实现逻辑：
- 显式将GCS Blob打开为标准类文件流，固定读写块大小为16MB（远低于2GB内存阈值）
- 压缩包使用顺序流式解压模式打开，禁止随机访问触发全量加载
- 解压后的内容逐块读取，按指定大小（比如单块256MB）切分后直接流式写回原存储桶
- 所有流传递环节均使用shutil.copyfileobj并显式传入length参数，控制单次内存拷贝的块大小

可直接运行的参考代码（以tar.gz格式压缩包为例，其他流式支持的压缩格式可调整解压mode参数）：

import shutil
import tarfile
from google.cloud import storage

# 配置参数
BUCKET_NAME = "你的存储桶名称"
SOURCE_BLOB_PATH = "源压缩包在桶内的路径，比如data/large_file.tar.gz"
CHUNK_SIZE = 16 * 1024 * 1024  # 单次读写块大小16MB，内存峰值不会超过这个值的2倍
SPLIT_BLOCK_SIZE = 256 * 1024 * 1024  # 解压后单个分块大小256MB
DEST_PREFIX = "unzip_split/"  # 解压分块在桶内的存储前缀

def main():
    # 初始化GCS客户端
    client = storage.Client()
    bucket = client.bucket(BUCKET_NAME)
    source_blob = bucket.blob(SOURCE_BLOB_PATH, chunk_size=CHUNK_SIZE)

    # 打开源压缩包为二进制只读流，全程不落地磁盘
    with source_blob.open("rb", chunk_size=CHUNK_SIZE) as source_stream:
        # 注意：tarfile必须用r|* 开头的流式模式，r:gz会触发随机seek导致全量加载
        with tarfile.open(fileobj=source_stream, mode="r|gz") as tar:
            block_index = 0
            current_block_written = 0
            current_block_stream = None

            for member in tar:
                # 跳过目录、软链，只处理普通文件
                if not member.isfile():
                    continue
                member_f = tar.extractfile(member)
                if not member_f:
                    continue
                
                # 逐块读取解压后的文件内容
                while True:
                    # 新建分块的逻辑：当前没有打开的上传流/当前分块写满
                    if current_block_stream is None or current_block_written >= SPLIT_BLOCK_SIZE:
                        if current_block_stream is not None:
                            # 关闭已写满的分块流，触发SDK完成上传
                            current_block_stream.close()
                        # 新建分块Blob
                        block_index += 1
                        dest_blob_path = f"{DEST_PREFIX}part_{block_index:06d}"
                        current_block_blob = bucket.blob(dest_blob_path, chunk_size=CHUNK_SIZE)
                        current_block_stream = current_block_blob.open("wb", chunk_size=CHUNK_SIZE)
                        current_block_written = 0
                    
                    # 用shutil.copyfileobj做低内存拷贝，单次最多读指定块大小
                    copy_length = min(CHUNK_SIZE, SPLIT_BLOCK_SIZE - current_block_written)
                    copied = shutil.copyfileobj(member_f, current_block_stream, length=copy_length)
                    if not copied:
                        break
                    current_block_written += copied
            
            # 关闭最后一个未写满的分块流
            if current_block_stream is not None:
                current_block_stream.close()

if __name__ == "__main__":
    main()

关键注意事项

如果你的压缩包是zip格式，不要直接用上面的流式解压逻辑：zip格式的文件索引存储在文件尾部，标准库zipfile必须做seek操作才能读取索引，这种场景下可以先读取压缩包最后10MB内容拿到中央目录偏移，再做范围读取解压，或者提前将zip包转换为tar.gz等支持顺序解压的格式
所有GCS流打开时必须显式传入chunk_size参数，否则SDK会默认使用100MB以上的默认块大小，极端情况下仍可能触发内存超限
shutil.copyfileobj的length参数必须显式传入，不传的话会默认读到缓冲区耗尽为止，大文件场景下会直接占满内存
不要调用download_as_string/download_to_filename/upload_from_string这类全量读写的API，这类API会把整个对象加载到内存后再处理，完全无法适配大文件场景

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