$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 または失敗に設定します。

sample_mflix データベースの movieArchive コレクションには、リリース年ごとの高評価映画の履歴レコードが格納されています。

movieArchive コレクションでは、year フィールドに一意のインデックスがあります。リリース年ごとに存在するレコードは最大 1 件でなければなりません。

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

この集計パイプラインは、movieArchive コレクションを movies コレクションのデータで更新し、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" } }
])

第 1 ステージ:

$match ステージ（コンテンツレーティングを持ち、1970 年から 1972 年の間にリリースされた、評価の高い映画（metacritic: 100）をフィルタリングします）。

第 2 ステージ:

$group ステージ（yearでグループ化し、新しい movieCount フィールドで映画をカウントします）。

第 3 ステージ:

$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" } }
])

第 1 ステージ:

$matchステージ（1970 年から 1972 年の間に投稿されたコメントをフィルタリングします）。

第 2 ステージ:

$group ステージ（コメント date から抽出した年でグループ化し、コメント数を新しい commentCount フィールドにカウントします）。

第 3 ステージ:

$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"
   } }
])

第 1 ステージ:

$match ステージ（1970 年 １ 月 15 日に投稿されたすべてのコメントを検索します）。

第 2 ステージ:

$group ステージ（一致したコメントを年月でグループ化し、その件数をカウントします）。

第 3 ステージ:

$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 のような形式です。

以下のタブは、変数を mergeステージ、aggregate コマンド、またはその両方で定義したときの動作を示します。

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' } ]