使用miniopy-async实现Minio对象批量打包上传的内存受限问题求助

核心思路 #

采用流式分块处理，全程不加载完整对象到内存，利用本地磁盘作为临时缓冲，完成ZIP归档后直接流式上传回Minio，同时控制临时文件大小避免占满磁盘。

具体实现步骤 #

1. 初始化Minio异步客户端 #

from miniopy_async import Minio
import asyncio

async def get_minio_client():
    client = Minio(
        "your-minio-endpoint",
        access_key="your-access-key",
        secret_key="your-secret-key",
        secure=True  # 按需调整是否启用HTTPS
    )
    return client

2. 流式生成ZIP归档（分块写入） #

用zipfile结合异步IO，每次从Minio流式下载对象分块，直接写入本地临时ZIP文件，全程不加载完整对象到内存：

import zipfile
import tempfile
from pathlib import Path

async def build_zip_stream(client, src_bucket, object_list, temp_zip_path):
    # 开启Zip64支持，兼容超过4GB的归档文件
    with zipfile.ZipFile(temp_zip_path, 'w', zipfile.ZIP_DEFLATED, allowZip64=True) as zipf:
        for obj_name in object_list:
            async with client.get_object(src_bucket, obj_name) as obj_stream:
                # 创建ZIP内的文件条目，保留原文件名
                zip_info = zipfile.ZipInfo(obj_name)
                zip_info.external_attr = 0o644 << 16  # 设置常规文件权限
                
                # 10MB分块读取写入，可根据内存情况微调
                chunk_size = 10 * 1024 * 1024
                with zipf.open(zip_info, 'w') as zip_entry:
                    while True:
                        chunk = await obj_stream.read(chunk_size)
                        if not chunk:
                            break
                        zip_entry.write(chunk)

3. 流式上传ZIP文件回Minio #

采用多部分上传，读取本地临时ZIP的分块直接上传，避免加载整个文件到内存：

async def upload_zip_stream(client, target_bucket, zip_obj_name, temp_zip_path):
    file_size = Path(temp_zip_path).stat().st_size
    chunk_size = 10 * 1024 * 1024  # 和下载分块保持一致
    
    # 初始化多部分上传
    upload_id = await client.create_multipart_upload(target_bucket, zip_obj_name)
    parts = []
    part_num = 1
    
    with open(temp_zip_path, 'rb') as f:
        while True:
            chunk = f.read(chunk_size)
            if not chunk:
                break
            # 上传单个分块
            etag = await client.upload_part(
                target_bucket, zip_obj_name, upload_id, part_num, chunk
            )
            parts.append({'PartNumber': part_num, 'ETag': etag})
            part_num += 1
    
    # 完成多部分上传
    await client.complete_multipart_upload(
        target_bucket, zip_obj_name, upload_id, parts
    )

4. 完整流程整合+临时文件清理 #

async def main():
    client = await get_minio_client()
    src_bucket = "your-source-bucket"
    target_bucket = "your-target-bucket"  # 可和源桶一致
    object_list = ["file1.dat", "file2.mp4", ...]  # 待打包的对象列表
    zip_obj_name = "archive-20240520.zip"
    
    # 创建临时ZIP文件，使用系统临时目录
    with tempfile.NamedTemporaryFile(suffix='.zip', delete=False) as tmp:
        temp_zip_path = tmp.name
    
    try:
        await build_zip_stream(client, src_bucket, object_list, temp_zip_path)
        await upload_zip_stream(client, target_bucket, zip_obj_name, temp_zip_path)
    finally:
        # 强制清理临时文件，避免磁盘占用
        Path(temp_zip_path).unlink(missing_ok=True)

if __name__ == "__main__":
    asyncio.run(main())

关键优化与限制处理 #

• 内存控制：分块大小设为10MB，单块内存占用极低，完全适配4GB内存限制。
• 磁盘容量适配：若待打包对象总压缩后大小超过240GB，可分批次处理——完成一批上传并清理临时文件后，再处理下一批。
• 压缩选项：若追求速度，可将ZIP_DEFLATED改为ZIP_STORED（无压缩），减少CPU开销。
• 容错处理：可给分块上传添加重试逻辑，应对网络波动导致的上传失败。

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