.NET 7下如何在AWS S3中打开可写流分批写入CSV文件？

问题核心 #

在.NET 7环境下，需实现从数据库通过EF分批读取百万级数据（每页1000行），将每批数据追加写入AWS S3的同一个CSV文件，同时最小化本地内存占用。原代码因流处理逻辑错误无法正常运行。

原代码问题分析 #

• 每次循环创建StreamWriter并使用using关键字，会直接关闭S3流，导致后续循环无法写入数据。
• TransferUtility.OpenStreamAsync默认以只读模式打开流，不支持写入操作。
• 使用Skip/Take分页处理超大规模数据集时，数据库需扫描大量前置数据，存在性能瓶颈。

修正方案 #

1. 正确处理S3写入流 #

调用TransferUtility.OpenStreamAsync时明确指定写入模式，确保流支持追加写入；将StreamWriter声明在循环外部，避免重复关闭底层流。

2. 优化EF分页（可选） #

针对百万级数据，推荐使用键值分页（基于主键或有序字段）替代Skip/Take，大幅提升数据库查询效率。

完整可运行代码示例 #

int pageSize = 1000;
long lastId = 0; // 假设数据表主键为long类型的Id，用于键值分页
bool hasMoreDocuments = true;

using var transferUtility = new TransferUtility(_s3Client);
// 以"不存在则创建、存在则追加"的模式打开S3文件流
using var s3Stream = await transferUtility.OpenStreamAsync(
    _bucketName, 
    path,
    FileMode.OpenOrCreate,
    FileAccess.Write);

// 将流指针移到末尾，避免覆盖已有内容
s3Stream.Seek(0, SeekOrigin.End);

// 在循环外部创建StreamWriter，设置leaveOpen: true避免关闭底层S3流
using var writer = new StreamWriter(s3Stream, Encoding.UTF8, leaveOpen: true);

// 写入CSV表头（仅执行一次）
await writer.WriteLineAsync(ConvertToCsvHeader());

while (hasMoreDocuments)
{
    // 键值分页：查询Id大于lastId的前pageSize条数据，替代低效的Skip/Take
    var pagedDocuments = await query
        .Where(doc => doc.Id > lastId)
        .OrderBy(doc => doc.Id)
        .Take(pageSize)
        .ToListAsync();

    if (!pagedDocuments.Any())
    {
        hasMoreDocuments = false;
        break;
    }

    foreach (var doc in pagedDocuments)
    {
        await writer.WriteLineAsync(ConvertToCsvRow(doc));
        lastId = doc.Id; // 更新最后一条数据的Id，用于下一页查询
    }

    // 立即刷新流，确保数据写入S3，避免内存堆积
    await writer.FlushAsync();
    await s3Stream.FlushAsync();

    hasMoreDocuments = pagedDocuments.Count == pageSize;
}

// 最终刷新所有缓存数据，确保全部写入S3
await writer.FlushAsync();
await s3Stream.FlushAsync();

关键改进点 #

• 流处理优化：通过指定FileMode.OpenOrCreate和FileAccess.Write实现写入权限，Seek到流末尾实现追加；StreamWriter设置leaveOpen: true避免关闭底层S3流。
• 内存控制：每次仅加载pageSize条数据到内存，写入后立即刷新，避免数据在本地堆积。
• 分页性能提升：键值分页避免了Skip/Take的全表扫描问题，数据库只需处理大于lastId的数据。

替代方案：S3分块上传（Multipart Upload） #

针对GB级超大型文件，推荐使用AWS S3的分块上传功能，每批数据写入独立内存块后上传，最后合并所有分块，进一步降低内存占用并提升可靠性。示例代码如下：

int pageSize = 1000;
long lastId = 0;
bool hasMoreDocuments = true;
List<UploadPartResponse> uploadParts = new();

// 初始化分块上传
var initiateRequest = new InitiateMultipartUploadRequest
{
    BucketName = _bucketName,
    Key = path
};
var initiateResponse = await _s3Client.InitiateMultipartUploadAsync(initiateRequest);

try
{
    int partNumber = 1;
    while (hasMoreDocuments)
    {
        var pagedDocuments = await query
            .Where(doc => doc.Id > lastId)
            .OrderBy(doc => doc.Id)
            .Take(pageSize)
            .ToListAsync();

        if (!pagedDocuments.Any())
        {
            hasMoreDocuments = false;
            break;
        }

        // 将当前批次数据写入内存流（仅占用单批次数据的内存）
        using var partStream = new MemoryStream();
        using var partWriter = new StreamWriter(partStream, Encoding.UTF8, leaveOpen: true);
        
        // 仅第一块写入CSV表头，避免重复
        if (partNumber == 1)
        {
            await partWriter.WriteLineAsync(ConvertToCsvHeader());
        }
        
        foreach (var doc in pagedDocuments)
        {
            await partWriter.WriteLineAsync(ConvertToCsvRow(doc));
            lastId = doc.Id;
        }

        await partWriter.FlushAsync();
        partStream.Seek(0, SeekOrigin.Begin);

        // 上传当前分块
        var uploadRequest = new UploadPartRequest
        {
            BucketName = _bucketName,
            Key = path,
            UploadId = initiateResponse.UploadId,
            PartNumber = partNumber,
            InputStream = partStream,
            PartSize = partStream.Length
        };

        var uploadResponse = await _s3Client.UploadPartAsync(uploadRequest);
        uploadParts.Add(uploadResponse);

        partNumber++;
        hasMoreDocuments = pagedDocuments.Count == pageSize;
    }

    // 合并所有分块，完成文件上传
    var completeRequest = new CompleteMultipartUploadRequest
    {
        BucketName = _bucketName,
        Key = path,
        UploadId = initiateResponse.UploadId,
        PartETags = uploadParts.Select(p => new PartETag(p.PartNumber, p.ETag)).ToList()
    };

    await _s3Client.CompleteMultipartUploadAsync(completeRequest);
}
catch (Exception ex)
{
    // 上传失败时中止分块上传，避免S3存储无效分块
    var abortRequest = new AbortMultipartUploadRequest
    {
        BucketName = _bucketName,
        Key = path,
        UploadId = initiateResponse.UploadId
    };
    await _s3Client.AbortMultipartUploadAsync(abortRequest);
    throw;
}

分块上传优势 #

• 无需长时间保持S3流连接，避免超时问题。
• 分块独立上传，失败后可仅重传失败块，提升可靠性。
• 内存占用更低，仅单批次数据驻留内存。

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