如何直接写入Azure存储并优化AppendBlob批量写入速度？

这是个非常典型的「内存占用」与「写入性能」的平衡问题，咱们直接拆解核心痛点，给出针对性的优化方案：

为什么你的AppendBlob写入速度会暴跌？ #

核心原因是每次调用AppendText都会发起一次独立的HTTP请求。如果你的reportEntries有几十万甚至上百万条数据，就意味着要发起几十万次网络请求——网络往返的开销（DNS解析、TCP握手、响应等待）直接把总耗时拉到了几小时级别。

优化方案按优先级排序 #

1. 批量拼接行，减少AppendBlob请求次数 #

这是成本最低、见效最快的优化：把多行数据拼接成一个大的字符串块，再一次性调用AppendText，把请求次数从N次降到N/批量大小，直接砍掉90%以上的网络开销。

示例代码：

public static void ExportCSVToStorageAccount(string fileName, CloudBlobContainer container, IEnumerable<RawReportRow> reportEntries)
{
    var blob = container.GetAppendBlobReference($"{fileName}.csv");
    blob.CreateOrReplace();
    // 先写入表头
    blob.AppendText($"Date,StoreId,ItemId,SalesQuantity{Environment.NewLine}");

    // 可根据单条数据大小调整批量值，建议每次批量100-1000条（对应10KB-100KB数据）
    const int batchSize = 500;
    var batchBuilder = new StringBuilder();
    int count = 0;

    foreach (var row in reportEntries)
    {
        batchBuilder.AppendLine($"\"{row.Date}\",\"{row.StoreId}\",\"{row.ItemId}\",\"{row.SalesQuantity}\"");
        count++;

        if (count % batchSize == 0)
        {
            blob.AppendText(batchBuilder.ToString());
            batchBuilder.Clear();
        }
    }

    // 写入最后一批剩余的数据
    if (batchBuilder.Length > 0)
    {
        blob.AppendText(batchBuilder.ToString());
    }
}

2. 改用异步API提升吞吐量 #

同步API会阻塞线程等待网络响应，改用异步API可以让线程在等待期间处理其他任务，尤其在数据量极大时能明显提升整体效率。

异步版本示例：

public static async Task ExportCSVToStorageAccountAsync(string fileName, CloudBlobContainer container, IEnumerable<RawReportRow> reportEntries)
{
    var blob = container.GetAppendBlobReference($"{fileName}.csv");
    await blob.CreateOrReplaceAsync();
    await blob.AppendTextAsync($"Date,StoreId,ItemId,SalesQuantity{Environment.NewLine}");

    const int batchSize = 500;
    var batchBuilder = new StringBuilder();
    int count = 0;

    foreach (var row in reportEntries)
    {
        batchBuilder.AppendLine($"\"{row.Date}\",\"{row.StoreId}\",\"{row.ItemId}\",\"{row.SalesQuantity}\"");
        count++;

        if (count % batchSize == 0)
        {
            await blob.AppendTextAsync(batchBuilder.ToString());
            batchBuilder.Clear();
        }
    }

    if (batchBuilder.Length > 0)
    {
        await blob.AppendTextAsync(batchBuilder.ToString());
    }
}

3. 若无需追加特性，改用BlockBlob流式上传 #

如果你的CSV文件写完后不需要再追加内容，推荐使用BlockBlob的流式上传——SDK会自动将流分成4MB大小的块进行上传，既避免内存过载，又能利用BlockBlob的高效分块传输优化，速度比AppendBlob更快。

示例代码（自定义流实现边生成边上传）：

public static async Task ExportCsvToBlockBlobAsync(string fileName, CloudBlobContainer container, IEnumerable<RawReportRow> reportEntries)
{
    var blockBlob = container.GetBlockBlobReference($"{fileName}.csv");
    // 边生成CSV内容边上传，无需加载全部数据到内存
    using (var csvStream = new CsvReportStream(reportEntries))
    {
        await blockBlob.UploadFromStreamAsync(csvStream);
    }
}

// 自定义流：从RawReportRow枚举中动态生成CSV内容
public class CsvReportStream : Stream
{
    private readonly IEnumerator<RawReportRow> _rowEnumerator;
    private readonly StringBuilder _lineBuilder = new StringBuilder();
    private byte[] _currentBuffer;
    private int _bufferPosition;

    public CsvReportStream(IEnumerable<RawReportRow> reportEntries)
    {
        _rowEnumerator = reportEntries.GetEnumerator();
        // 先写入表头
        _lineBuilder.AppendLine("Date,StoreId,ItemId,SalesQuantity");
        _currentBuffer = Encoding.UTF8.GetBytes(_lineBuilder.ToString());
        _lineBuilder.Clear();
    }

    // 核心Read方法：按需生成CSV行的字节数据
    public override int Read(byte[] buffer, int offset, int count)
    {
        if (_currentBuffer == null || _bufferPosition >= _currentBuffer.Length)
        {
            if (!_rowEnumerator.MoveNext())
            {
                return 0; // 无更多数据
            }
            var row = _rowEnumerator.Current;
            _lineBuilder.AppendLine($"\"{row.Date}\",\"{row.StoreId}\",\"{row.ItemId}\",\"{row.SalesQuantity}\"");
            _currentBuffer = Encoding.UTF8.GetBytes(_lineBuilder.ToString());
            _lineBuilder.Clear();
            _bufferPosition = 0;
        }

        var bytesToCopy = Math.Min(count, _currentBuffer.Length - _bufferPosition);
        Array.Copy(_currentBuffer, _bufferPosition, buffer, offset, bytesToCopy);
        _bufferPosition += bytesToCopy;
        return bytesToCopy;
    }

    // 实现Stream的其他必要抽象方法
    public override bool CanRead => true;
    public override bool CanSeek => false;
    public override bool CanWrite => false;
    public override long Length => throw new NotSupportedException();
    public override long Position { get => throw new NotSupportedException(); set => throw new NotSupportedException(); }
    public override void Flush() => throw new NotSupportedException();
    public override long Seek(long offset, SeekOrigin origin) => throw new NotSupportedException();
    public override void SetLength(long value) => throw new NotSupportedException();
    public override void Write(byte[] buffer, int offset, int count) => throw new NotSupportedException();

    protected override void Dispose(bool disposing)
    {
        if (disposing)
        {
            _rowEnumerator.Dispose();
        }
        base.Dispose(disposing);
    }
}

4. 升级到最新Azure存储SDK #

如果你还在使用旧版Microsoft.WindowsAzure.Storage包，强烈建议升级到Azure.Storage.Blobs（Azure SDK v12+）。新SDK有更高效的HTTP客户端、优化的重试策略和性能提升，API设计也更简洁。

新SDK下的AppendBlob批量写入示例：

using Azure.Storage.Blobs.Specialized;

public static async Task ExportCSVToStorageAccountAsync(string fileName, BlobContainerClient containerClient, IEnumerable<RawReportRow> reportEntries)
{
    var appendBlobClient = containerClient.GetAppendBlobClient($"{fileName}.csv");
    await appendBlobClient.CreateIfNotExistsAsync();
    await appendBlobClient.AppendTextAsync($"Date,StoreId,ItemId,SalesQuantity{Environment.NewLine}");

    const int batchSize = 500;
    var batchBuilder = new StringBuilder();
    int count = 0;

    foreach (var row in reportEntries)
    {
        batchBuilder.AppendLine($"\"{row.Date}\",\"{row.StoreId}\",\"{row.ItemId}\",\"{row.SalesQuantity}\"");
        count++;

        if (count % batchSize == 0)
        {
            await appendBlobClient.AppendTextAsync(batchBuilder.ToString());
            batchBuilder.Clear();
        }
    }

    if (batchBuilder.Length > 0)
    {
        await appendBlobClient.AppendTextAsync(batchBuilder.ToString());
    }
}

5. 额外优化建议 #

• 确保你的应用与Azure存储账户在同一区域，减少网络延迟；
• 配置SDK的重试策略（比如指数退避），避免临时网络问题拖慢速度；
• 若数据量超大（10GB+），可以考虑并行生成块数据并上传BlockBlob（注意BlockBlob的块顺序）。

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