如何在Rust中实现从S3直接解压缩文件并解析（无需存盘）

要实现从S3下载后直接解压缩并解析（无需写入磁盘），核心是利用Rust的异步流式IO能力，将S3返回的ByteStream直接转换为可读取的异步流，再通过异步压缩库解压缩，最后流式解析JSON。以下是具体实现方案：

1. 调整依赖 #

首先在Cargo.toml中添加所需依赖：

[dependencies]
anyhow = "1.0"
aws-sdk-s3 = "0.31"
tokio = { version = "1.0", features = ["full"] }
tokio-stream = "0.1"
async-compression = { version = "0.4", features = ["tokio", "gzip"] }
tokio-util = { version = "0.7", features = ["io"] }
serde = { version = "1.0", features = ["derive"] }
serde_json = "1.0"

2. 修改核心逻辑（流式处理） #

替换原有的磁盘写入逻辑，改为直接在内存中流式处理：

const ENV_CRED_KEY_ID: &str = "KEY_ID";
const ENV_CRED_KEY_SECRET: &str = "KEY_SECRET";
const BUCKET_NAME: &str = "bucketname";
const REGION: &str = "us-east-1";

use anyhow::{anyhow, bail, Context, Result};
use aws_sdk_s3::{config, ByteStream, Client, Credentials, Region};
use async_compression::tokio::bufread::GzDecoder;
use serde::Deserialize;
use std::env;
use tokio_stream::StreamExt;
use tokio_util::io::StreamReader;

// 定义你的JSON数据结构（根据实际业务调整字段）
#[derive(Debug, Deserialize)]
struct CellData {
    id: u64,
    cell_value: String,
    record_time: i64,
}

#[tokio::main]
async fn main() -> Result<()> {
    let client = get_aws_client(REGION)?;

    let keys = list_keys(&client, BUCKET_NAME, "CELLDATA/year=2022/month=06/day=06/").await?;
    println!("List:\n{}", keys.join("\n"));

    let key: &str = &keys[0];
    parse_s3_gz_json(&client, BUCKET_NAME, key).await?;

    Ok(())
}

async fn parse_s3_gz_json(client: &Client, bucket_name: &str, key: &str) -> Result<()> {
    // 发起S3请求并获取响应流
    let res = client.get_object().bucket(bucket_name).key(key).send().await?;
    let byte_stream: ByteStream = res.body;

    // 将ByteStream转换为AsyncRead兼容的异步读取器
    let stream_reader = StreamReader::new(byte_stream);
    // 包裹为Gzip解码器，实现流式解压缩
    let mut decoder = GzDecoder::new(stream_reader);

    // 流式解析JSON，逐条处理数据（避免加载整个大文件到内存）
    let mut deserializer = serde_json::Deserializer::from_reader(&mut decoder);
    while let Some(data) = CellData::deserialize(&mut deserializer).transpose()? {
        // 这里添加业务逻辑处理，比如写入数据库、计算统计值等
        println!("解析到数据: {:?}", data);
    }

    Ok(())
}

fn get_aws_client(region: &str) -> Result<Client> {
    let key_id = env::var(ENV_CRED_KEY_ID).context("Missing S3_KEY_ID")?;
    let key_secret = env::var(ENV_CRED_KEY_SECRET).context("Missing S3_KEY_SECRET")?;

    let cred = Credentials::new(key_id, key_secret, None, None, "loaded-from-custom-env");
    let region = Region::new(region.to_string());
    let conf_builder = config::Builder::new().region(region).credentials_provider(cred);
    let conf = conf_builder.build();

    Ok(Client::from_conf(conf))
}

// 补充实现list_keys函数（原代码未提供）
async fn list_keys(client: &Client, bucket_name: &str, prefix: &str) -> Result<Vec<String>> {
    let mut keys = Vec::new();
    let mut response = client.list_objects_v2().bucket(bucket_name).prefix(prefix).send().await?;

    while let Some(obj_list) = response.contents.take() {
        for item in obj_list {
            if let Some(key) = item.key {
                keys.push(key);
            }
        }

        if let Some(token) = response.next_continuation_token.take() {
            response = client
                .list_objects_v2()
                .bucket(bucket_name)
                .prefix(prefix)
                .continuation_token(token)
                .send()
                .await?;
        } else {
            break;
        }
    }

    Ok(keys)
}

关键逻辑说明 #

• ByteStream转AsyncRead：通过tokio_util::io::StreamReader将S3返回的字节流转换为异步读取器，让压缩库可以直接读取流数据。
• 异步流式解压缩：使用async-compression的tokio版本，支持异步环境下的流式解压缩，避免阻塞线程。
• 内存级JSON解析：serde_json::Deserializer支持从读取器流式解析JSON，即使处理GB级文件也不会占用过多内存，逐条处理解析后的结构体。
• 完全移除磁盘操作：全程无需创建文件、写入磁盘，所有数据都在内存流中完成处理，和你的Go代码逻辑完全对齐。

注意事项 #

• 必须确保CellData结构体的字段与实际JSON文件的键名匹配，否则解析会失败。
• 如果S3中的压缩文件是其他格式（如deflate），可替换GzDecoder为对应格式的解码器。
• 生产环境中可替换anyhow为thiserror自定义错误类型，让错误处理更精准。

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