.NET中无需将大文件存入内存的下载上传方案咨询

针对1GB级Gzip压缩CSV的流式处理需求，核心思路是构建下载-解压-修改-上传的全流式管道，全程不加载完整文件到内存，以下是具体实现方案：

核心原则 #

• 所有环节采用流操作，避免一次性读取整个文件
• 按批次处理并写入数据，平衡内存占用与IO效率
• 利用语言原生的流式处理工具和云端SDK的流式上传接口

Python 实现示例 #

方案说明 #

用gzip模块流式解压，csv模块逐行读写，结合云端SDK的流式上传接口（以AWS S3为例），通过自定义流对象实现处理后数据的实时传递。

import gzip
import csv
from io import BytesIO
import boto3

def process_row(row):
    # 自定义行修改逻辑，示例：将第一列转为大写
    row[0] = row[0].upper()
    return row

class ProcessedCSVStream:
    def __init__(self, download_stream):
        self.gz_stream = gzip.GzipFile(fileobj=download_stream, mode='rb')
        self.csv_reader = csv.reader((line.decode('utf-8') for line in self.gz_stream))
        self.header_written = False
        self.buffer = b""

    def read(self, size=-1):
        # 先写入表头
        if not self.header_written:
            header = next(self.csv_reader)
            processed_header = process_row(header)
            self.buffer += ','.join(processed_header).encode('utf-8') + b'\n'
            self.header_written = True
        
        # 填充缓冲区直到达到指定大小或文件结束
        while size == -1 or len(self.buffer) < size:
            try:
                row = next(self.csv_reader)
                processed_row = process_row(row)
                self.buffer += ','.join(processed_row).encode('utf-8') + b'\n'
            except StopIteration:
                break
        
        # 返回指定大小的数据
        if size == -1:
            data = self.buffer
            self.buffer = b""
            return data
        else:
            data = self.buffer[:size]
            self.buffer = self.buffer[size:]
            return data

# 调用示例
if __name__ == "__main__":
    s3 = boto3.client('s3')
    # 获取云端下载流
    download_stream = s3.get_object(Bucket="your-source-bucket", Key="input.gz")["Body"]
    # 流式上传处理后的数据
    processed_stream = ProcessedCSVStream(download_stream)
    s3.upload_fileobj(processed_stream, "your-target-bucket", "output.csv")

Java 实现示例 #

方案说明 #

用GZIPInputStream流式解压，Apache Commons CSV库逐行解析修改，结合云端SDK的流式上传接口（以阿里云OSS为例），通过批次缓冲区控制内存占用。

import org.apache.commons.csv.CSVFormat;
import org.apache.commons.csv.CSVParser;
import org.apache.commons.csv.CSVPrinter;
import org.apache.commons.csv.CSVRecord;

import java.io.*;
import java.util.ArrayList;
import java.util.List;

public class CloudCSVProcessor {
    private static final int BATCH_SIZE = 1000; // 每批次处理行数

    // 自定义行修改逻辑
    private static List<String> processRow(List<String> row) {
        row.set(0, row.get(0).toUpperCase());
        return row;
    }

    public static void processAndUpload(InputStream downloadStream, OutputStream uploadStream) throws IOException {
        try (GZIPInputStream gzipIn = new GZIPInputStream(downloadStream);
             BufferedReader reader = new BufferedReader(new InputStreamReader(gzipIn));
             CSVParser csvParser = new CSVParser(reader, CSVFormat.DEFAULT.withHeader());
             CSVPrinter csvPrinter = new CSVPrinter(new OutputStreamWriter(uploadStream), CSVFormat.DEFAULT)) {

            // 写入表头
            csvPrinter.printRecord(csvParser.getHeaderNames());
            csvPrinter.flush();

            List<List<String>> batchBuffer = new ArrayList<>(BATCH_SIZE);
            int rowCount = 0;

            for (CSVRecord record : csvParser) {
                List<String> processedRow = processRow(new ArrayList<>(record.toList()));
                batchBuffer.add(processedRow);
                rowCount++;

                // 达到批次大小则写入并刷新
                if (rowCount >= BATCH_SIZE) {
                    for (List<String> row : batchBuffer) {
                        csvPrinter.printRecord(row);
                    }
                    uploadStream.flush();
                    batchBuffer.clear();
                    rowCount = 0;
                }
            }

            // 处理剩余数据
            if (!batchBuffer.isEmpty()) {
                for (List<String> row : batchBuffer) {
                    csvPrinter.printRecord(row);
                }
                uploadStream.flush();
            }
        }
    }

    // 调用示例
    public static void main(String[] args) throws IOException {
        com.aliyun.oss.OSS ossClient = com.aliyun.oss.OSSClientBuilder.create().build();
        // 获取云端下载流
        InputStream downloadStream = ossClient.getObject("your-source-bucket", "input.gz").getObjectContent();
        // 获取云端上传流
        OutputStream uploadStream = ossClient.putObject("your-target-bucket", "output.csv", new ByteArrayInputStream(new byte[0])).getOutputStream();

        processAndUpload(downloadStream, uploadStream);
        ossClient.shutdown();
    }
}

关键注意事项 #

• 流的生命周期管理：用语言的自动资源管理机制（Python上下文管理器、Java try-with-resources）确保流正确关闭，避免资源泄漏。
• 批次大小调整：根据应用内存限制调整批次大小，一般1000-10000行可兼顾内存占用与IO效率。
• 云端SDK适配：确认使用的云端存储SDK支持流式上传（如AWS S3 upload_fileobj、阿里云OSS流式putObject），超大文件可结合分块上传接口进一步优化。

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