c# - 什么是带有事务的 mongo 4.4 批量大小？

Question

我的问题是当我使用 session 在 bulkInsert 操作中插入 ~10k 文档时我收到错误

{"Command insert failed: WriteConflict error: this operation conflicted with another operation. Please retry your operation or multi-document transaction.."}

但是，当我插入 ~9,5k 或更少的相同文档时，一切正常。我的 mongo 集群托管在 AWS 云中。
我只发现单个 BsonDocument 最大大小为 16mb 的信息。我如何处理更大的批量插入？

var bulkOps = new List<WriteModel<BsonDocument>>();  
... some code here..  
var upsertOne = new InsertOneModel<BsonDocument>(bsonDoc);

bulkOps.Add(upsertOne);

return collection.BulkWrite(session, bulkOps);

Answer 1

我在使用 BulkWriteAsync 时遇到了类似的问题与 ReplaceOneModel (你使用 BulkWrite ...我建议你去异步，但这是一个旁白)。
我有一个执行良好的策略，可以处理大量插入的大量文档。我认为我的方法与首先制作小批量的方法相结合可能会奏效(引用 this answer)。该战略由一些战术组成......
策略一:使用IsOrdered
通过BulkWriteOptions至 BulkWriteAsync IsOrdered 设置为 false . IsOrdered允许 MongoDB 执行插入 faster .更重要的是，在您的情况下，它可能允许 many documents 的操作成功。 (不一定是全部——见策略 2)。
注意:我提供的最后一个链接是 Java 文档，它说，

If true, then when a write fails, return without performing the remaining writes.

C# 文档 talks仅关于订购本身。无论如何，这个使用IsOrdered为假设置第二个策略。
策略2:当批量插入失败时，重试剩余的
问题是，对大量文档的完整批量插入可能会失败——但不会完全失败。这实际上是可以的，因为 MongoDB 给了你一个很好的结果:作为 MongoBulkWriteException 的一部分无法写入(插入)的文档列表。那是抛出。同样，我的场景是一个 upsert，但我愿意打赌严格的 insert 会有相同/相似的问题。
使用 WriteErrors此异常的属性，您可以获取这些文档的列表，然后重试它们。
可能的解决方案
在这里，我概述了一个可能的解决方案。我不打算提供整套代码，因为我不能提供，而且我还有其他一些策略，例如使用 exponential backoff 重试。和 jitter ，它是通用存储库实现的一部分，我处理各种其他错误，等等 - 可能与此答案相关也可能不相关的事情。
因此，使用类型为 T 的文档的伪代码:

// setup models; here's mine; yours will be `InsertOneModel`
var models = toUpdate
    .Select(x => new ReplaceOneModel<T>(new ExpressionFilterDefinition<T>(doc => doc.Id == x.Id), x) { IsUpsert = true })
    .ToArray();

然后进行重试循环并处理必要的异常，这会暴露 WriteErrors类型的属性 IReadOnlyList<BulkWriteError> .您必须对我的代码做出一些假设；在错误处理程序中，我必须将错误与原始模型相匹配，以便确保我拥有正确的模型。如果您不明白，我可以尝试添加更多上下文。

for (var i = 0; ; i++)
{
    try
    {
        await collection.BulkWriteAsync(models, new BulkWriteOptions { IsOrdered = false });
        return result;
    }
    catch (MongoBulkWriteException<T> bwe)
    {
        if (i > DelayCount) return something;

        // rebuild the collection of models, using the failed ones; EntityModel
        // basically contains the entity, and the error category (you may not
        // want to retry all of them depending on the category)
        var myModels = bwe.WriteErrors.Select(x => new EntityModel<T> { Entity = models[x.Index].Replacement, ErrorCategory = (RepoErrorCategory)x.Category }).ToArray();
        models = handlerResult.Replacements
            .Select(x => new ReplaceOneModel<T>(new ExpressionFilterDefinition<T>(doc => doc.Id == x.Id), x) { IsUpsert = upsert })
            .ToArray();
         
        // maybe do an optional delay here...
    }
}

将错误与模型匹配的关键是使用 Index WriteError 的属性(property)在错误列表中查找您尝试保存的原始实体(文档)。
在这一点上，在尽可能少的迭代之后，一切都应该被插入，并且已经绕过了大小和时间限制。
本质上，在每次通过时，您每次都将剩余的工作筛选出失败的文档。理想情况下，每次都更少。
最终建议
如果可以的话，我建议不要进行这种交易。在这种情况下，这些策略可能无法正常工作。例如，很明显，10000 个文档的事务性插入将无法使用 WriteErrors。 .我敢打赌，在大多数情况下，10000 个文档是不相关的，它们的插入可以是独立的。
避免交易将使您更容易绕过限制。