如何将MongoDB的clicks集合增量合并至sessions集合？

需求说明 #

将 clicks 集合的点击数据合并到 sessions 集合，规则如下：

• 已有会话：累加 clicks 计数，保留原有 start_time
• 新会话：添加新文档，start_time 取该会话最早的点击时间，clicks 计数为1
• 支持增量执行：仅处理 time 大于上次处理最大时间的点击数据，避免重复操作

1. 单次合并操作 #

先对 clicks 按 session_id 分组统计，再批量更新 sessions：

// 聚合 clicks 数据，得到每个会话的点击量和最早时间
const clickStats = db.clicks.aggregate([
  // 首次执行可移除该过滤条件，增量执行时替换为上次处理的最大时间
  { $match: { time: { $gt: 0 } } },
  {
    $group: {
      _id: "$session_id",
      clickCount: { $sum: 1 },
      firstClickTime: { $min: "$time" }
    }
  }
]).toArray();

// 构建批量更新操作
const bulkUpdates = clickStats.map(stat => ({
  updateOne: {
    filter: { session_id: stat._id },
    update: {
      $inc: { clicks: stat.clickCount },
      // 仅当会话不存在时设置 start_time
      $setOnInsert: { start_time: stat.firstClickTime }
    },
    upsert: true
  }
}));

// 执行批量操作
if (bulkUpdates.length > 0) {
  db.sessions.bulkWrite(bulkUpdates);
}

2. 增量执行方案 #

为避免重复处理，维护一个记录上次处理时间的配置文档：

// 获取上次处理的最大时间（存储在 process_metadata 集合）
const lastProcessRecord = db.process_metadata.findOne({ name: "click_session_merge" });
const lastProcessedTime = lastProcessRecord ? lastProcessRecord.max_time : 0;

// 聚合待处理的 clicks 数据，同时获取本次处理的最大时间
const aggregateResult = db.clicks.aggregate([
  { $match: { time: { $gt: lastProcessedTime } } },
  {
    $group: {
      _id: "$session_id",
      clickCount: { $sum: 1 },
      firstClickTime: { $min: "$time" }
    }
  },
  {
    $group: {
      _id: null,
      maxTime: { $max: "$firstClickTime" },
      sessionStats: { $push: "$$ROOT" }
    }
  }
]).toArray()[0];

if (aggregateResult && aggregateResult.sessionStats.length > 0) {
  // 执行批量更新
  const bulkOps = aggregateResult.sessionStats.map(stat => ({
    updateOne: {
      filter: { session_id: stat._id },
      update: {
        $inc: { clicks: stat.clickCount },
        $setOnInsert: { start_time: stat.firstClickTime }
      },
      upsert: true
    }
  }));
  db.sessions.bulkWrite(bulkOps);

  // 更新上次处理时间记录
  db.process_metadata.updateOne(
    { name: "click_session_merge" },
    { $set: { max_time: aggregateResult.maxTime } },
    { upsert: true }
  );
}

3. 验证结果 #

初始数据 #

clicks 集合：

{"session_id": 2, "time": 12}, 
{"session_id": 2, "time": 12.1}, 
{"session_id": 3, "time": 13}

sessions 集合：

{"session_id":1, "start_time": 1, "clicks": 1},
{"session_id":2, "start_time": 2, "clicks": 1}

合并后 sessions 集合 #

{"session_id": 1, "start_time": 1, "clicks": 1},  // 原有数据无变化
{"session_id": 2, "start_time": 2, "clicks": 3}, // clicks 累加为 1+2=3，start_time 保留原值
{"session_id": 3, "start_time": 13, "clicks": 1} // 新增会话，start_time 取最早点击时间

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