$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 movieRatingSummarysample_mflix数据库中的 集合

要查看新 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，则聚合不会影响匹配文档，也不会出现重复键错误。同样，如果指定替换，操作不会失败；但是，该操作会替换现有文档。

合并多个集合的结果

默认下，如果聚合结果中的文档与集合中的文档匹配，则$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" } }
])

第一个阶段：

$matchmetacritic: 100阶段筛选内容分级在1970 和 之间上映的广受好评的电影 ()。1972

第二阶段：

$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 之间发布的评论。

第二阶段：

$groupdatecommentCount阶段按从评论 中提取的年份进行分组，并将评论计数到新的 字段中。

第三阶段：

$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阶段会查找在 1 月15 发布的所有评论。1970

第二阶段：

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

第三阶段：

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

• 此管道无法直接访问权限结果文档中的字段。为了访问权限结果文档中的 count字段，管道使用 $$new 变量；即 $$new.count。

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

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

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

db.monthlyCommentTotals.find()

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

使用变量自定义合并

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

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

• $merge stage 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' } ]