如何用MongoDB的$merge实现周聚合数据的增量更新

我有一个存储近百万款产品每日销量数据的集合，已经通过以下聚合查询生成了周度聚合数据集合：

[
  {
    $match:
      {
        $and: [
          {
            "_id.date": {
              $gte: "20220103",
            },
          },
          {
            "_id.date": {
              $lte: "20230122",
            },
          },
        ],
      },
  },
  {
    $project:
      {
        _id: 1,
        realDate: {
          $dateFromString: {
            dateString: "$_id.date",
            format: "%Y%m%d",
          },
        },
        count: 1,
      },
  },
  {
    $group:
      {
        _id: {
          name: "$_id.name",
          year: {
            $isoWeekYear: "$realDate",
          },
          week: {
            $isoWeek: "$realDate",
          },
        },
        total: {
          $sum: "$count",
        },
      },
  },
  {
    $group:
      {
        _id: "$_id.name",
        dates: {
          $addToSet: {
            year: "$_id.year",
            week: "$_id.week",
            count: "$total",
          },
        },
      },
  },
  {
    $merge:
      {
        into: "dataPerWeek",
        on: "_id",
      },
  },
]

该查询生成的文档结构如下：

{
  "_id": "myProduct",
  "dates": [
    {
      "year": {
        "$numberLong": "2022"
      },
      "week": 52,
      "count": 10
    },
    {
      "year": {
        "$numberLong": "2022"
      },
      "week": 50,
      "count": 6
    },
    {
      "year": {
        "$numberLong": "2022"
      },
      "week": 49,
      "count": 2
    },
    {
      "year": {
        "$numberLong": "2022"
      },
      "week": 51,
      "count": 5
    },
    {
      "year": {
        "$numberLong": "2023"
      },
      "week": 1,
      "count": 5
    },
    {
      "year": {
        "$numberLong": "2023"
      },
      "week": 2,
      "count": 2
    },
    {
      "year": {
        "$numberLong": "2023"
      },
      "week": 3,
      "count": 4
    }
  ]
}

现在我希望每周更新这个集合，仅向dates数组中新增当周的销量数据（若产品文档不存在则新建），但当前重复执行上述$merge查询会覆盖原有数据，请问能否通过单条查询实现这种增量更新？

可以通过修改聚合查询中的$merge阶段，指定whenMatched的更新逻辑，实现仅新增当周数据的增量更新，不会覆盖原有dates数组。

具体修改如下：

1. 调整$match阶段，仅筛选当周的日期范围（格式为YYYYMMDD），避免重复处理历史数据；
2. 修改$merge阶段的whenMatched参数，通过条件判断仅添加不存在的周数据，whenNotMatched保持默认的insert即可。

修改后的完整聚合查询：

[
  {
    $match:
      {
        "_id.date": {
          // 替换为当周的日期范围，示例：2024年第20周（20240513-20240519）
          $gte: "20240513",
          $lte: "20240519"
        }
      },
  },
  {
    $project:
      {
        _id: 1,
        realDate: {
          $dateFromString: {
            dateString: "$_id.date",
            format: "%Y%m%d",
          },
        },
        count: 1,
      },
  },
  {
    $group:
      {
        _id: {
          name: "$_id.name",
          year: {
            $isoWeekYear: "$realDate",
          },
          week: {
            $isoWeek: "$realDate",
          },
        },
        total: {
          $sum: "$count",
        },
      },
  },
  {
    $group:
      {
        _id: "$_id.name",
        newDate: {
          $first: {
            year: "$_id.year",
            week: "$_id.week",
            count: "$total",
          },
        },
      },
  },
  {
    $merge:
      {
        into: "dataPerWeek",
        on: "_id",
        whenMatched: [
          {
            $addFields: {
              dates: {
                $cond: [
                  {
                    $anyElementTrue: [
                      {
                        $map: {
                          input: "$dates",
                          as: "d",
                          in: {
                            $and: [
                              { $eq: ["$$d.year", "$newDate.year"] },
                              { $eq: ["$$d.week", "$newDate.week"] }
                            ]
                          }
                        }
                      }
                    ]
                  },
                  // 该周数据已存在，保持原dates数组不变
                  "$dates",
                  // 该周数据不存在，追加新数据到数组
                  { $concatArrays: ["$dates", ["$newDate"]] }
                ]
              }
            }
          },
          // 清理临时字段newDate
          { $unset: "newDate" }
        ],
        whenNotMatched: "insert"
      },
  },
]

关键逻辑说明： #

• 第二个$group阶段改用$first获取当前周的单条数据（因为已按产品+年+周分组，每个产品只会有一条当周数据）；
• $merge的whenMatched逻辑中，通过$map和$anyElementTrue判断当前周数据是否已存在于dates数组：
  • 若存在，保留原数组；
  • 若不存在，用$concatArrays将新数据追加到数组末尾；
• 最后用$unset清理临时字段newDate，避免污染文档结构。

每次执行该查询时，只会新增当周的销量数据，不会覆盖原有历史数据；若产品文档不存在，则自动创建新文档并添加当周数据。

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