Python环境下动态生成数据直接流式上传至S3的实现方案咨询

该需求完全可实现，无需将数据写入本地磁盘。boto3的S3上传接口原生支持类文件对象作为输入，不需要传入本地文件路径，你只需要将动态生成的数据流封装为符合Python可读协议的对象即可直接上传。

核心思路是自定义实现一个可读流类，将异步队列中生成的有序数据块喂给该流，直接传入upload_fileobj方法完成串行流式上传。boto3会自动根据你配置的块大小处理上传逻辑，无需手动管理分片。

具体实现步骤 #

• 第一步：实现自定义流式读取类，仅需要实现read方法适配boto3的读取逻辑：

import asyncio
import boto3
from boto3.s3.transfer import TransferConfig
from botocore.exceptions import ClientError

class QueueStreamReader:
    def __init__(self, queue: asyncio.Queue, max_block_size: int, done_signal: object):
        self.queue = queue
        self.max_block_size = max_block_size
        self.done_signal = done_signal
        self._buffer = b""

    def read(self, n: int = -1) -> bytes:
        read_target = n if n != -1 else self.max_block_size
        # 缓冲不足时从队列拉取数据
        while len(self._buffer) < read_target:
            item = self.queue.get()
            if item is self.done_signal:
                # 所有数据生产完成，返回剩余缓冲
                res = self._buffer
                self._buffer = b""
                return res
            self._buffer += item
        # 截取指定长度返回，剩余数据留到下一次读取
        res = self._buffer[:read_target]
        self._buffer = self._buffer[read_target:]
        return res

• 第二步：串联异步数据生产流程和S3上传流程，设置串行上传参数：

async def main():
    # 自定义配置
    MAX_BLOCK_SIZE = 10 * 1024 * 1024  # 10MB，需大于所有单个数据单元的大小
    S3_BUCKET = "你的存储桶名称"
    S3_OBJECT_KEY = "上传到S3的文件路径"
    QUEUE_DONE_SIGNAL = object()  # 数据生产完成的标记信号

    # 初始化队列、流对象、S3客户端
    data_queue = asyncio.Queue(maxsize=10)  # 设置队列大小做背压控制，避免内存溢出
    stream_reader = QueueStreamReader(data_queue, MAX_BLOCK_SIZE, QUEUE_DONE_SIGNAL)
    s3_client = boto3.client("s3")

    # 你的数据生产协程，替换为实际的异步队列数据处理逻辑
    async def data_producer():
        # 示例：模拟生成100个大小不一的数据单元
        for idx in range(100):
            data_unit = f"测试数据单元_{idx}".encode("utf-8")
            await data_queue.put(data_unit)
        # 所有数据生产完成，写入结束信号
        await data_queue.put(QUEUE_DONE_SIGNAL)

    producer_task = asyncio.create_task(data_producer())

    # 在线程池中运行同步上传逻辑，避免阻塞异步事件循环
    loop = asyncio.get_running_loop()
    await loop.run_in_executor(
        None,
        lambda: s3_client.upload_fileobj(
            Fileobj=stream_reader,
            Bucket=S3_BUCKET,
            Key=S3_OBJECT_KEY,
            Config=TransferConfig(
                multipart_threshold=MAX_BLOCK_SIZE,
                multipart_chunksize=MAX_BLOCK_SIZE,
                max_concurrency=1,  # 强制串行上传，无并发
                use_threads=False
            )
        )
    )

    await producer_task

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

• 由于你的数据单元是无序生成的，需要确保写入队列之前已经完成了数据的排序拼接，保证喂给流对象的字节顺序和最终文件的预期顺序一致，否则上传后的文件内容会错乱。
• 可根据服务器内存配置调整队列的最大长度，做背压控制，避免数据生产速度远快于上传速度时内存占用过高。
• 可根据需要自行添加上传失败重试、异常捕获和资源释放的相关逻辑。

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