Flask API流式传输S3对象时如何对发送至客户端的数据流进行压缩

问题1 流式字节流压缩方案 #

你可以采用流式gzip压缩实现边拉取S3分块、边压缩、边返回给客户端，无需加载完整文件到内存，具体实现方式如下：

• 首先校验客户端请求头的Accept-Encoding字段，确认客户端支持gzip压缩，避免返回客户端无法解码的内容
• 使用Python标准库zlib的compressobj对象（配置wbits=zlib.MAX_WBITS | 16即可生成标准gzip格式压缩流），对每次从S3拉取的分块做增量压缩
• 构造Flask响应时设置正确的响应头：Content-Encoding: gzip、Transfer-Encoding: chunked、Vary: Accept-Encoding，不要设置Content-Length头，因为流式压缩后的最终大小无法提前预知
• 如果不想自己实现压缩逻辑，也可以使用Flask扩展flask-compress，只需将配置项COMPRESS_STREAMS设为True，即可自动处理流式响应的gzip压缩

示例代码如下：

import zlib
from flask import Flask, Response, request

app = Flask(__name__)
# S3拉取分块大小可自行调整，建议设置为8MB~16MB平衡性能和内存占用
CHUNK_SIZE = 8 * 1024 * 1024

def get_s3_file_size(bucket: str, key: str) -> int:
    # 先调用S3 head接口获取文件总大小，用于构造Range请求
    resp = aws.head_object(Bucket=bucket, Key=key)
    return resp['ContentLength']

def stream_compressed_s3(bucket: str, key: str, total_size: int):
    compress_obj = zlib.compressobj(wbits=zlib.MAX_WBITS | 16)
    remaining_bytes = total_size
    while remaining_bytes > 0:
        start = total_size - remaining_bytes
        end = min(start + CHUNK_SIZE - 1, total_size - 1)
        s3_resp = aws.get_object(
            Bucket=bucket,
            Key=key,
            Range=f"bytes={start}-{end}"
        )
        chunk = s3_resp['Body'].read()
        remaining_bytes -= len(chunk)
        compressed_chunk = compress_obj.compress(chunk)
        if compressed_chunk:
            yield compressed_chunk
    # 输出压缩器剩余的缓存数据
    yield compress_obj.flush()

@app.route("/s3-json")
def s3_json_proxy():
    bucket, key = 解析传入的S3参数逻辑()
    total_size = get_s3_file_size(bucket, key)
    accept_encoding = request.headers.get("Accept-Encoding", "").lower()
    if "gzip" in accept_encoding:
        resp = Response(
            stream_compressed_s3(bucket, key, total_size),
            content_type="application/json"
        )
        resp.headers["Content-Encoding"] = "gzip"
        resp.headers["Transfer-Encoding"] = "chunked"
        resp.headers["Vary"] = "Accept-Encoding"
        return resp
    else:
        # 客户端不支持gzip时返回未压缩的流式响应
        def stream_raw_s3():
            remaining_bytes = total_size
            while remaining_bytes > 0:
                start = total_size - remaining_bytes
                end = min(start + CHUNK_SIZE - 1, total_size - 1)
                s3_resp = aws.get_object(
                    Bucket=bucket,
                    Key=key,
                    Range=f"bytes={start}-{end}"
                )
                chunk = s3_resp['Body'].read()
                remaining_bytes -= len(chunk)
                yield chunk
        return Response(stream_raw_s3(), content_type="application/json")

问题2 gzip压缩是否需要预加载完整文件 #

不需要，gzip底层基于DEFLATE算法，本身支持流式增量压缩，只有使用gzip.compress()这类一次性压缩接口时才需要传入完整文件内容。你使用zlib.compressobj或者gzip.GzipFile这类支持增量写入的对象时，完全可以每拿到一段S3分块就压缩一段，整个过程服务端内存占用仅和你设置的单分块大小有关，即使是GB级别的文件也不会出现内存溢出问题。

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