.NET7下MongoDB 7时间序列集合批量Upsert性能优化求助

我在.NET 7环境中使用MongoDB 7及MongoDB.Driver 2.24.0，现有一个MongoDB时间序列集合，仅执行插入操作时运行正常，但现在会出现重复记录，需要实现插入/更新（Upsert）逻辑。我搜索了Upsert相关示例，但发现其与时间序列集合不兼容，于是采用了先批量删除重复记录再批量插入的实现方式。

实体类定义 #

public class Ctimeseries {
    [BsonId]
    [BsonRepresentation(BsonType.ObjectId)]
    public string? Id { get; set; }
    /// <summary>
    /// Date in format "2024-03-12T14:14:57.000Z"
    /// </summary>
    public DateTime timestamp { get; set; }// = new (0,0);
    /// <summary>
    /// string with an alfanumeric code
    /// </summary>
    public Metadata metadata { get; set; } = new Metadata();
    /// <summary>
    /// set of 8 measurements
    /// </summary>
    public double[] Measurements { get; set; } 
    /// <summary>
    /// Summer Flag
    /// </summary>
    public byte Summer { get; set; }
    /// <summary>
    /// kind of measure, default 1
    /// </summary>
    public byte MeasureKind { get; set; } = 1;
    /// <summary>
    /// extra data 26 caracters
    /// </summary>
    public string Extradata { get; set; }
}

当前Upsert实现代码 #

public async Task Upsert(List<Ctimeseries> records)
{
    //queue all the deletes from incoming records and bulk delete/insert
    List<DeleteOneModel<Ctimeseries>> requests = new List<DeleteOneModel<Ctimeseries>>(records.Count());
    foreach (Ctimeseries record in records) {
        var filter = new FilterDefinitionBuilder<Ctimeseries>().Where(m => m.timestamp == record.timestamp && m.metadata == record.metadata);
        var request = new DeleteOneModel<Ctimeseries>(filter); //queue the delete
        requests.Add(request);
    }
    await _CtimeseriesCol.BulkWriteAsync(requests); //perform bulk delete <- "flowControl": {"acquireCount": 2232},"durationMillis": 3748    1,67921146953405 ms per record
    await _CtimeseriesCol.InsertManyAsync(records); //perform bulk insert <- "flowControl": {"acquireCount": 54133},"durationMillis": 92724  1,712892320765522 ms per record
}

性能问题与报错 #

该实现性能表现极差：使用纯InsertManyAsync可在约5分钟内插入900万条记录，而当前方案运行1小时仅处理了100多万条记录。此前尝试参考批量Upsert示例，但报错：

A bulk write operation resulted in one or more errors. WriteErrors: [ { Category : "Uncategorized", Code : 72, Message : "Cannot perform a non-multi update on a time-series collection" } ]

请问我是否做了多余的操作？是否需要通过Bson文档调用C#驱动来实现更优的方案？

核心问题分析 #

你当前的方案性能差的原因在于：

• 批量删除时每条记录都生成单独的DeleteOneModel，相当于执行N次单条删除操作，MongoDB需要反复处理过滤条件和锁资源，开销极大。
• 先删后插的逻辑需要两次全量批量操作，额外增加了网络IO和数据库写入开销。

另外，时间序列集合不支持常规Upsert（报错Code72）是因为MongoDB限制了对时间序列集合的更新操作——仅允许多文档更新，且不能修改timestamp和metadata字段（这两个是时间序列集合的主键字段）。

优化方案 #

1. 合并删除操作，减少批量请求数量

不要为每条记录单独生成DeleteOneModel，而是将相同过滤条件的记录合并，使用DeleteManyModel批量删除重复项：

public async Task UpsertOptimized(List<Ctimeseries> records)
{
    // 提取所有需要去重的(timestamp, metadata)对，避免重复过滤
    var uniqueKeys = records.Select(r => new { r.timestamp, r.metadata }).Distinct().ToList();
    
    // 构建批量删除请求
    var deleteRequests = new List<WriteModel<Ctimeseries>>();
    foreach (var key in uniqueKeys)
    {
        var filter = Builders<Ctimeseries>.Filter.Where(m => 
            m.timestamp == key.timestamp && m.metadata == key.metadata);
        deleteRequests.Add(new DeleteManyModel<Ctimeseries>(filter));
    }
    
    // 执行批量删除，设置无序执行提高性能
    if (deleteRequests.Any())
    {
        await _CtimeseriesCol.BulkWriteAsync(deleteRequests, new BulkWriteOptions { IsOrdered = false });
    }
    
    // 执行批量插入
    await _CtimeseriesCol.InsertManyAsync(records, new InsertManyOptions { IsOrdered = false });
}

2. 利用BulkWrite合并删除与插入操作

将删除和插入操作合并到同一个BulkWrite请求中，减少网络往返次数：

public async Task UpsertBulkCombined(List<Ctimeseries> records)
{
    var bulkOperations = new List<WriteModel<Ctimeseries>>();
    var processedKeys = new HashSet<string>();
    
    foreach (var record in records)
    {
        // 生成唯一键哈希值，避免重复处理相同的(timestamp, metadata)对
        var keyHash = $"{record.timestamp:O}|{JsonSerializer.Serialize(record.metadata)}";
        if (processedKeys.Add(keyHash))
        {
            var filter = Builders<Ctimeseries>.Filter.Where(m => 
                m.timestamp == record.timestamp && m.metadata == record.metadata);
            bulkOperations.Add(new DeleteManyModel<Ctimeseries>(filter));
        }
        // 添加插入操作
        bulkOperations.Add(new InsertOneModel<Ctimeseries>(record));
    }
    
    // 执行批量操作，无序执行提升速度
    await _CtimeseriesCol.BulkWriteAsync(bulkOperations, new BulkWriteOptions { IsOrdered = false });
}

3. 关键性能优化点

• 设置IsOrdered = false：允许MongoDB并行执行操作，大幅提升批量处理速度（前提是操作之间无依赖）。
• 去重过滤条件：避免对相同的(timestamp, metadata)对重复执行删除操作，减少无效请求。
• 合并操作类型：将删除和插入合并到一个BulkWrite请求中，减少网络IO次数。

关于Bson文档调用的说明 #

不需要刻意改用Bson文档调用C#驱动，上述优化方案使用强类型的Builders和WriteModel已经足够高效。如果需要更底层的控制，可以将实体转为BsonDocument，但性能提升有限，反而会增加代码复杂度。

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