$merge（聚合阶段）

按需物化视图：初始创建

如果输出集合不存在，则 $merge 会创建该集合。

你可以使用 $group 和 $merge 阶段来创建一个 movieRatingSummary 集合，用于按上映年份和内容分级汇总获得高度评价的影片：

db.movies.aggregate( [
   { $match: { metacritic: 100, rated: { $ne: null },
     year: { $lte: 1972 } } },
   { $group: { _id: { year: "$year", rated: "$rated" },
     count: { $sum: 1 } } },
   { $merge : { into: "movieRatingSummary", on: "_id",
     whenMatched: "replace", whenNotMatched: "insert" } }
] )

该管道使用以下阶段：

• $match 用于筛选通过 1972 年发行且具有内容评级的广受好评电影的阶段

• $group 阶段以按 year 和 rated 对这些影片进行分组

• $merge 阶段将前一个 $group 阶段的输出写入 sample_mflix 数据库中的 movieRatingSummary 集合。

要查看新 movieRatingSummary 集合中的文档：

db.movieRatingSummary.find().sort(
   { _id: 1 } )

[
  { _id: { year: 1939, rated: 'PASSED' }, count: 1 },
  { _id: { year: 1962, rated: 'PG' }, count: 1 },
  { _id: { year: 1963, rated: 'PG' }, count: 1 },
  { _id: { year: 1970, rated: 'R' }, count: 1 },
  { _id: { year: 1972, rated: 'R' }, count: 1 }
]

按需物化视图：更新/替换数据

要更新上一个示例中的 movieRatingSummary 集合，使其包含从 1963 年起获得高度评价的影片，该聚合管道使用以下阶段：

• $match 阶段用于查找所有包含 metacritic: 100（内容分级）且上映年份晚于或等于 1963 的影片。

• $group 阶段以按 year 和 rated 对这些影片进行分组。

• $merge 将结果集写入 movieRatingSummary 集合，并用替换文档中具有相同 _id 值的记录。对于集合中没有匹配项的文档，$merge 会插入新文档。

db.movies.aggregate( [
   { $match: { metacritic: 100, rated: { $ne: null },
     year: { $gte: 1963 } } },
   { $group: { _id: { year: "$year", rated: "$rated" },
     count: { $sum: 1 } } },
   { $merge : { into: "movieRatingSummary", on: "_id",
     whenMatched: "replace", whenNotMatched: "insert" } }
] )

聚合运行结束后，查看movieRatingSummary集合中的文档：

db.movieRatingSummary.find().sort(
   { _id: 1 } )

[
  { _id: { year: 1939, rated: 'PASSED' }, count: 1 },
  { _id: { year: 1962, rated: 'PG' }, count: 1 },
  { _id: { year: 1963, rated: 'PG' }, count: 1 },
  { _id: { year: 1970, rated: 'R' }, count: 1 },
  { _id: { year: 1972, rated: 'R' }, count: 1 },
  { _id: { year: 1982, rated: 'R' }, count: 1 },
  { _id: { year: 2014, rated: 'R' }, count: 1 }
]

仅插入新数据

为确保 $merge 不会覆盖集合中的现有数据，请将 whenMatched 设置为 keepExisting 或 fail。

sample_mflix 数据库中的集合 movieArchive 包含各发行年份备受好评电影的历史记录。

movieArchive 集合具有针对 year 字段的唯一索引。每个发布年份至多有一条记录：

db.movieArchive.createIndex(
   { year: 1 }, { unique: true } )

此聚合管道使用 movies 集合中的数据更新 movieArchive 集合，以包含从 1963 年开始以来广受好评的电影。该管道使用以下阶段：

• $match 阶段查找所有具有 metacritic: 100、内容分级和 year >= 1963 的电影。

• $group 阶段按 year 对电影标题进行分组。

• $project 阶段以抑制 _id 字段并将 year 提升到顶级字段。将文档传递给 $merge 时，$merge 会自动为这些文档生成一个新的 _id 字段。

• $merge 将结果集写入 movieArchive。

  $merge 阶段会在 year 字段上匹配文档，并在匹配时触发失败。也就是说，如果某个发行年份的文档已经存在，$merge 将会报错。

db.movies.aggregate( [
   { $match: { metacritic: 100, rated: { $ne: null },
     year: { $gte: 1963 } } },
   { $group: { _id: "$year",
     titles: { $push: "$title" } } },
   { $project: { _id: 0, year: "$_id", titles: 1 } },
   { $merge : { into: "movieArchive", on: "year",
     whenMatched: "fail" } }
] )

如果 movieArchive 集合中已存在该范围 1963 年到 2014 年期间任意年份的文档，则该聚合会因重复键错误而失败。然而，该管道不会回滚在发生错误之前已插入的任何文档。

如果为匹配文档指定 keepExisting，则聚合不会影响匹配文档，也不会出现重复键错误。同样，如果指定 replace，操作不会失败，但会替换现有文档。

合并多个集合的结果

默认情况下，如果聚合结果中的文档与集合中的文档相匹配，$merge 阶段就会合并文档。

您可以使用 $merge 来合并来自 movies 集合和 comments 集合的结果，从而创建一个新的集合 yearlyStats。

要创建 yearlyStats 集合，请运行以下管道：

db.movies.aggregate( [
   { $match: { metacritic: 100, rated: { $ne: null },
     year: { $gte: 1970, $lte: 1972 } } },
   { $group: { _id: "$year", movieCount: { $sum: 1 } } },
   { $merge : { into: "yearlyStats", on: "_id",
     whenMatched: "merge", whenNotMatched: "insert" } }
])

第一个阶段：

$match 阶段筛选出具有内容分级、在 1970 年到 1972 年期间发布的广受好评的电影 (metacritic: 100)。

第二阶段：

$group 阶段按 year 分组，并将电影计入新的 movieCount 字段。

第三阶段：

$merge 阶段会将这些文档写入同一数据库中的 yearlyStats 集合。如果此阶段在集合中找到与 _id 字段匹配的现有文档，此阶段则会合并这些匹配的文档。否则，此阶段会插入该文档。对于初始创建，所有文档都不应匹配。

要查看集合中的文档，运行以下操作：

db.yearlyStats.find().sort( { _id: 1 } )

[
  { _id: 1970, movieCount: 1 },
  { _id: 1972, movieCount: 1 }
]

同样，针对 comments 集合运行以下聚合管道，将评论计数合并到 yearlyStats 集合中。

db.comments.aggregate( [
   { $match: { date: { $gte: new Date("1970-01-01"),
     $lt: new Date("1973-01-01") } } },
   { $group: { _id: { $year: "$date" },
     commentCount: { $sum: 1 } } },
   { $merge : { into: "yearlyStats", on: "_id",
     whenMatched: "merge", whenNotMatched: "insert" } }
])

第一个阶段：

$match 阶段：筛选 1970 年到 1972 年之间发布的评论。

第二阶段：

$group 阶段按从注释 date 中提取的年份进行分组，并将注释计数到新的 commentCount 字段中。

第三阶段：

$merge 阶段将这些文档写入同一数据库中的 yearlyStats 集合。如果该阶段在集合中找到与 _id 字段（年度）匹配的现有文档，则该阶段会合并匹配文档。否则，该阶段会插入文档。

要在合并数据后查看 yearlyStats 集合中的文档，请运行以下操作：

db.yearlyStats.find().sort( { _id: 1 } )

[
  { _id: 1970, movieCount: 1, commentCount: 889 },
  { _id: 1971, commentCount: 825 },
  { _id: 1972, movieCount: 1, commentCount: 863 }
]

使用管道自定义合并

当文档匹配时，$merge 可以使用自定义更新管道。whenMatched 管道可以包含以下阶段：

• $addFields 及其别名 $set

• $project 及其别名 $unset

• $replaceRoot 及其别名 $replaceWith

集合 monthlyCommentTotals 追踪每月评论的累计数量。

sample_mflix.comments 集合每天都有新评论。以下聚合管道使用当天的评论计数更新每月总数：

db.comments.aggregate([
   { $match: { date: { $gte: new Date("1970-01-15"),
     $lt: new Date("1970-01-16") } } },
   { $group: { _id: { $dateToString: { format: "%Y-%m",
     date: "$date" } }, count: { $sum: 1 } } },
   { $merge: {
         into: "monthlyCommentTotals",
         on: "_id",
         whenMatched: [
            { $addFields: {
               count: { $add: [ "$count", "$$new.count" ] }
            } }
         ],
         whenNotMatched: "insert"
   } }
])

第一个阶段：

$match 阶段可查找发布于 1970 年 1 月 15 日的所有评论。

第二阶段：

$group 阶段按年月对匹配的评论进行分组并对其进行计数。

第三阶段：

$merge 阶段将文档写入 monthlyCommentTotals 集合。如果该阶段在集合中找到与 _id 字段匹配的现有文档，则该阶段会使用管道将当天的 count 添加到现有的月度总数中。

• 该管道不能直接访问结果文档中的字段。要访问结果文档中的 count 字段，管道使用 $$new 变量；即 $$new.count。

• 此管道可直接访问集合中现有文档的 count 字段；即 $count。

生成的文档将替换现有文档。

要在合并操作后查看 monthlyCommentTotals 集合中的文档，请运行以下操作：

db.monthlyCommentTotals.find()

[ { _id: '1970-01', count: 71 } ]

使用变量自定义合并

您可以在 $merge 阶段 whenMatched 字段中使用变量。必须先定义变量，然后才能使用。

使用以下一种或两种方式定义变量：

• $merge 阶段 let

• aggregate 命令 let（从 MongoDB 5.0 开始）

要在 whenMatched 中使用变量，请执行以下操作：

以 $$<variable_name> 形式指定双美元符号 ($$) 前缀以及变量名称。例如，$$year。如果将此变量设为文档，则还能以 $$<variable_name>.<field> 形式包含文档字段。例如，$$year.month。

下面的标签页展示在合并阶段、聚合命令或两者中定义变量时的行为。

db.movies.aggregate( [
   { $match: { title: "The Godfather" } },
   { $limit: 1 },
   { $project: { title: 1 } },
   { $merge: { into: "movieDetails", whenNotMatched: "insert" } }
] )
db.runCommand( {
   aggregate: db.movieDetails.getName(),
   pipeline: [ {
      $merge: {
         into: db.movieDetails.getName(),
         let : { year: "2023" },
         whenMatched: [ {
            $addFields: { "addedYear": "$$year" }
         } ]
      }
   } ],
   cursor: {}
} )
db.movieDetails.find()

[ { _id: ..., title: 'The Godfather', addedYear: '2023' } ]

db.movies.aggregate( [
   { $match: { title: "The Godfather" } },
   { $limit: 1 },
   { $project: { title: 1 } },
   { $merge: { into: "movieDetails", whenNotMatched: "insert" } }
] )
db.runCommand( {
   aggregate: db.movieDetails.getName(),
   pipeline: [ {
      $merge: {
         into: db.movieDetails.getName(),
         whenMatched: [ {
            $addFields: { "addedYear": "$$year" }
         } ]
      }
   } ],
   cursor: {},
   let : { year: "2023" }
} )
db.movieDetails.find()

[ { _id: ..., title: 'The Godfather', addedYear: '2023' } ]

db.movies.aggregate( [
   { $match: { title: "The Godfather" } },
   { $limit: 1 },
   { $project: { title: 1 } },
   { $merge: { into: "movieDetails", whenNotMatched: "insert" } }
] )
db.runCommand( {
   aggregate: db.movieDetails.getName(),
   pipeline: [ {
      $merge: {
         into: db.movieDetails.getName(),
         let : { year: "2023" },
         whenMatched: [ {
            $addFields: { "addedYear": "$$year" }
         } ]
      }
   } ],
   cursor: {},
   let : { year: "2019" }
} )
db.movieDetails.find()

[ { _id: ..., title: 'The Godfather', addedYear: '2023' } ]