如何用Python流式上传至AWS S3？Lambda处理S3 Zip文件遇空文件问题

嘿，我来帮你拆解这两个问题，一步步解决你的困惑——不管是流式上传S3的基础思路，还是你Lambda代码里的空文件和流式验证问题，咱们逐个理清楚。

一、Python流式上传到AWS S3的核心逻辑 #

流式上传的关键就是别把整个文件一股脑塞进内存，而是分块读取、分块上传。boto3的upload_fileobj方法天生适合这个场景，它能直接接受类文件对象（比如网络流、S3的流式Body、本地文件句柄），自动帮你分块处理，不用自己写复杂的分块逻辑。

举个简单的本地流上传例子：

import boto3
import io

s3 = boto3.client('s3')
# 假设你有一个生成内容的流（比如从数据库导出的数据流）
with io.BytesIO() as stream:
    # 模拟分块写入内容
    for chunk in your_content_generator():
        stream.write(chunk)
    stream.seek(0)  # 一定要把指针移回开头，不然上传的是空内容！
    s3.upload_fileobj(stream, '你的桶名', '目标文件路径')

二、你的Lambda代码问题修复与流式优化 #

先说说你当前代码的两个核心问题：

1. 空文件的锅：你写完out_buffer后，文件指针停在内容最后面，直接调用upload_fileobj的话，从这个位置读不到任何内容，自然生成空文件。解决办法就是上传前执行out_buffer.seek(0)重置指针。
2. 假流式的坑：obj.get()["Body"].read()这行代码会把整个Zip文件一次性加载到内存的buffer里，完全没做到流式处理——如果Zip文件很大，Lambda的内存直接就爆了。

修正后的真正流式代码 #

下面的代码会真正做到按需读取Zip内容，不一次性加载整个文件，同时解决空文件问题：

import boto3
import zipfile

s3 = boto3.resource('s3')
bucket_name = '你的桶名'
key = '源Zip文件的S3路径'

obj = s3.Object(bucket_name, key)
# 直接用S3的流式Body创建ZipFile，不用一次性读全量内容
with zipfile.ZipFile(obj.get()["Body"]) as z:
    for file_info in z.filelist:
        # 跳过Zip里的目录（如果有的话）
        if file_info.is_dir():
            continue
        # 直接打开Zip内的单个文件，返回流式对象
        with z.open(file_info) as zip_file_stream:
            target_key = f"output/{file_info.filename}_output"
            # 直接把流式对象传给upload_fileobj，一步到位
            s3.Object(bucket_name, target_key).upload_fileobj(zip_file_stream)
            print(f"搞定！已上传: {target_key}")

关键优化点解释 #

• 真·流式处理：zipfile.ZipFile可以直接接受S3的StreamingBody（它本身就是类文件对象），ZipFile会按需从S3读取Zip的目录和单个文件内容，不会把整个Zip塞进内存，完美适配Lambda的内存限制。
• 解决空文件：直接用zip_file_stream上传，upload_fileobj会从指针起始位置开始读，不需要手动重置；如果一定要用out_buffer中转，记得上传前out_buffer.seek(0)。
• 代码更简洁：用with语句自动管理资源，不用手动调用close()，减少出错概率。

怎么验证是不是真的流式？ #

有两个简单方法：

1. 看Lambda监控：去AWS控制台看Lambda的执行日志和内存使用指标，如果内存占用稳定在一个较低值，不随Zip文件大小涨，那就是流式；如果内存占用接近Zip文件大小，说明还是一次性加载了整个文件。
2. 加调试日志：在循环里打印每次读取的块大小，比如：

with z.open(file_info) as zip_file_stream:
    chunk_size = 1024 * 1024  # 每次读1MB
    total_bytes = 0
    while True:
        chunk = zip_file_stream.read(chunk_size)
        if not chunk:
            break
        total_bytes += len(chunk)
        print(f"刚读了 {len(chunk)} 字节，累计 {total_bytes} 字节")
    # 然后再上传

如果看到日志里分块打印的内容，就说明是流式读取没错。

三、额外提醒 #

• Lambda临时磁盘：咱们的代码没用到磁盘，所以不用担心Lambda默认的512MB临时空间限制；如果后续有需要写磁盘的场景，记得注意这个限制。
• 权限要给够：确保Lambda的执行角色有S3的GetObject（读源Zip）和PutObject（传解压后的文件）权限，不然会报错。

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