Azure CosmosDB+Azure Function批量插入2700万条数据的优化咨询

分批调用AddBulkAsync的可行性 #

这个方案完全可行，但需要注意单批次任务数量的控制和并发度限制：

• 你当前每个batch的8.2-8.5万条记录，如果直接全量丢给Task.WhenAll，依然可能导致单批次内存占用过高、CosmosDB请求突增触发限流（429错误）。建议把每个batch再拆成更小的子批次（比如每1万条一个子批次），每次只执行一个子批次的Task.WhenAll，或者用信号量控制同时运行的任务数量。
• 多次调用Task.WhenAll本身不会有问题，只要你不是同时启动所有batch的任务，而是串行或有限并行地处理每个batch，就能避免瞬间打满资源。

更优实现方式 #

1. 使用CosmosDB原生批量操作API #

不要自己用CreateItemAsync+Task.WhenAll实现批量，改用CosmosDB SDK提供的原生批量操作，它会自动优化请求分批、重试和资源占用，比手动创建大量任务更高效，还能减少429错误：

var bulkOperations = new List<Operation>();
foreach (var item in batchItems)
{
    bulkOperations.Add(Operation.CreateItem(item));
}
await container.ExecuteBulkOperationsAsync(bulkOperations);

2. 严格控制并发与批次大小 #

• 每个子批次的大小建议控制在1000-5000条之间（根据单条记录大小调整，单条越大，批次越小），避免单批次内存过载。
• 用SemaphoreSlim限制并发插入任务数，防止同时发起过多请求导致CosmosDB限流或Function内存溢出：

var semaphore = new SemaphoreSlim(100);
var tasks = new List<Task>();
foreach (var item in batchItems)
{
    await semaphore.WaitAsync();
    tasks.Add(container.CreateItemAsync(item).ContinueWith(t => semaphore.Release()));
}
await Task.WhenAll(tasks);

3. 优化Azure Function配置 #

• 如果用消耗计划，调整函数实例内存到更高规格（比如2GB），避免内存不足导致实例重启；专用计划则确保实例数量和资源足够支撑批量任务。
• 临时关闭函数的动态缩放，或设置最大实例数，防止短时间内大量实例启动打满CosmosDB。
• 调整函数超时时间，批量插入耗时较长，默认超时可能不够，需在host.json配置：

{
  "functionTimeout": "01:00:00"
}

4. 优化CosmosDB配置与策略 #

• 确保分区键选择合理，如果batchId本身是分区键，按batch处理天然能分散请求到不同分区，避免热点限流。
• 临时提高CosmosDB的吞吐量（RU/s），批量完成后再降回去，减少429错误影响；用自动缩放的话，确保阈值足够高。

5. 增加重试与监控机制 #

• 对CosmosDB的429错误实现指数退避重试，可调整SDK默认重试策略适配批量场景。
• 记录每个批次的插入结果，失败批次单独保存重试，避免全量任务失败重跑。
• 监控CosmosDB的请求速率、RU使用率和429错误率，实时调整批次大小和并发度。

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