使用 insertMany 的 CosmosDb 请求率很大答案

【问题标题】：CosmosDb Request rate is large with insertMany使用 insertMany 的 CosmosDb 请求率很大
【发布时间】：2019-08-11 07:58:03
【问题描述】：

我有以下存储库类在 CosmosDb 数据库中批量插入数据：

public bool InsertZonierData(List<Zonier> zonierList)
{
    if (zonierList == null || !zonierList.Any())
    {
        throw new ZonierListNullOrEmptyException();
    }
    else
    {
        try
        {
            _collection.InsertMany(zonierList);
            return true;
        }
        catch (MongoBulkWriteException ex)
        {
            throw new DataBaseWritingException(ex.Message, ExceptionCodeConstants.DataBaseWritingExceptionCode);
        }
    }
}

不幸的是，zonierList 中有超过 30000 个元素，它在 CosmosDb 上引发以下异常：

未处理的异常：MongoDB.Driver.MongoCommandException：命令插入失败：消息：{“错误”：[“请求率很大”]}

根据文档，这是与 Cosmos 上的 RU / sec 相关的问题。当然，一个简单的方法是增加它，但这不是我想做的。

是否有一种简单明了的方法来重构该方法，允许我们在不破坏 CosmosDb 的 400 RU/秒的情况下插入数据。

【问题讨论】：

标签： c# mongodb azure azure-cosmosdb

【解决方案1】：

Mongo 驱动程序会告诉您哪些记录出错，哪些记录根本没有处理。如果所有错误（通常是一个）的代码为 16500，那么您的问题是限制并重试错误，并且剩余记录是安全的。否则你的错误是由其他原因引起的，你应该分析并决定是否继续重试。

Mongo 驱动程序不返回 HTTP 标头，其中 Cosmos DB 建议在重试之前延迟，但这没什么大不了的。无论如何，延迟并不能保证成功，因为其他访问同一数据库的请求可能会用完 RU。您最好尝试并确定自己的重试规则。下面是一个简单的递归解决方案，它会不断重试，直到一切正常或达到重试限制。

    private async Task InsertManyWithRetry(IMongoCollection<BsonDocument> collection, 
        IEnumerable<BsonDocument> batch, int retries = 10, int delay = 300)
    {
        var batchArray = batch.ToArray();

        try
        {
            await collection.InsertManyAsync(batchArray);
        }
        catch (MongoBulkWriteException<BsonDocument> e)
        {
            if (retries <= 0)
                throw;

            //Check if there were any errors other than throttling.
            var realErrors = e.WriteErrors.Where(we => we.Code != 16500).ToArray();
            //Re-throw original exception for now.
            //TODO: We can make it more sophisticated by continuing with unprocessed records and collecting all errors from all retries.
            if (realErrors.Any())
                throw;

            //Take all records that had errors.
            var errors = e.WriteErrors.Select(we => batchArray[we.Index]);
            //Take all unprocessed records.
            var unprocessed = e.UnprocessedRequests
                .Where(ur => ur.ModelType == WriteModelType.InsertOne)
                .OfType<InsertOneModel<BsonDocument>>() 
                .Select(ur => ur.Document);

            var retryBatchArray = errors.Union(unprocessed).ToArray();

            _logger($"Retry {retryBatchArray.Length} records after {delay} ms");

            await Task.Delay(delay);

            await InsertManyWithRetry(collection, retryBatchArray, retries - 1, delay);
        }
    }

【讨论】：

【解决方案2】：

mongo sdk 完全不知道 CosmosDB 的存在。这意味着它对受限制的请求没有任何重试逻辑。这意味着，如果您想将 RU 保持在 400，则必须批量处理您的列表并使用客户端限制机制调用 insertmany 方法。

您可以通过获取每个文档的大小来计算，将其乘以 10，即 1kb 文档的插入费用，然后编写一段代码，根据大小对文档进行批处理并每秒执行一次。

【讨论】：

我同意这可能是目前使用 MongoDB 接口处理它的唯一方法。微软确实引入了某种批量操作库来抽象出这种逻辑，但不支持这个接口。问题是任何数据库操作都可能导致速率异常，具体取决于当时发生的其他情况。重试单个插入或获取或写入很容易，但对于 InsertMany 来说，麻烦在于制作逻辑和额外的数据库调用以了解实际创建了哪些记录以及需要重试哪些记录。使用 IsOrdered = true 可能会有所帮助。

【解决方案3】：

我通过使用 mongo api 对 cosmos bs 使用重试逻辑解决了这个问题。您可以根据自己的要求申请延迟。

public void Insert(List<BsonDocument> list)
    {
        try
        {
            var collection = this.db.GetCollection<BsonDocument>(COLLECTION_NAME);
            collection.InsertMany(list);
        } catch (MongoBulkWriteException ex)
        {
            int index = ex.WriteErrors[0].Index;
            Insert(list.GetRange(index, list.Count - index));
        }

    }

【讨论】：