ストリーミングマテリアライズドビューをビルドする

ストリーミングマテリアライズドビューは、Atlas Stream Processing ストリームプロセッサーが更新し続けるコレクションです。これを実現するため、ストリームプロセッサーはソースコレクションへの変更を読み取り、その変更の影響を計算し、その結果をビューに適用します。ビューは、手動またはスケジュールされた更新なしでソースデータの現在の状態を反映します。

ストリーミングマテリアライズドビューは、ダッシュボード、実行中の合計、開始アイテムのカウントなど、ソースデータの変更よりも計算結果の読み取りが多いワークロードに適しています。データが到着するとビューが更新されるため、読み取りにより、低レイテンシで現在の結果が返されます。

オンデマンドマテリアライズドビューとの比較

MongoDB は、マテリアライズドビューに対する2つのアプローチをサポートしています。

オンデマンドのマテリアライズドビューには、手動または予定に基づいて実行する集計パイプラインの結果が保存されます。
ストリーミングマテリアライズドビューは、継続的に実行されるAtlas Stream Processingストリームプロセッサの結果を保存します。

どの方法でも、計算結果をディスクに保存し、ビューから直接読み取りを行います。ビューの更新方法と時期が異なります。

特性	オンデマンドのマテリアライズドビュー	ストリーミングマテリアライズドビュー
trigger の更新	手動またはスケジュールされたもの	変更駆動型、継続的
レイテンシ	数分から数日	1 秒未満～数秒
データの新鮮度	点インタイムスナップショット	永久に同期される
コンピュートモデル	全体の結果を再計算します	増分効果を計算します
Best fit	バッチするレポート作成、定期的集計	ライブダッシュボード、運用分析

ストリーミングマテリアライズドビュープロセッサの特性

ストリーミングマテリアライズドビューの維持は、標準の Atlas Stream Processing 集計ステージから構成されるパスワードなしです。このパスワードなしを実装するストリームプロセッサには、次の特徴があります。

変更ストリームソースを読み取ります。fullDocument と fullDocumentBeforeChange が required に設定された $source ステージは、パイプラインが変更前後の各ドキュメントの状態を比較できるように、ソースコレクションから読み取ります。
イベントごとに符号付きの差分を計算します。$addFields ステージでは、$switch 式を使用して、変更が計算結果にどのように影響するかに基づいて、各挿入、更新、または削除に正または負の値を割り当てることができます。
結果をウィンドウにグループ化します。ストリームプロセッサーは無限のストリームで動作するため、すべての $group ステージはウィンドウステージ内で実行する必要があります。ウィンドウ化された $group は、ウィンドウ間隔でキーごとの差分を合計します。ウィンドウ間隔はビューの更新間隔としても機能するため、ビューの新鮮度を決定します。設定できる最小の間隔は 1 ミリ秒です。
結果を追加して適用します。$merge ステージでは、各ウィンドウの結果をビューの実行中の合計に追加するパイプライン whenMatched を使用できます。これにより、結果が置き換えられることはありません。
シンクステージで終了します。Stream Processing パイプラインはシンクステージで終了する必要があります。Use $merge を使用して Atlas コレクションに書き込み (write)ます。

増分集計のみがストリーミングマテリアライズドビューに変換されます。完全なコレクションスキャンが必要な集計は対象外です。

考慮する必要がある動作の違い

ストリーミングのマテリアライズドビューは、ダウンストリームのコンシューマーに影響を与える方法でバッチ集計とは異なります。

ビューはゼロから始まります。デフォルトでは、プロセッサは既存のドキュメントを読み取らないため、プロセッサを開始する前にビューをシードするか、$source ステージで initialSync を有効にします。
ソースの変更のみが更新を行います。パイプラインで $lookup ステージを使用する場合、参照コレクションに後で変更が加えられても、ビューが既に書き込み (write) を行ったドキュメントは更新されません。

ストリーミングマテリアライズドビューの作成

次のチュートリアルでは、購入方法別の完了した販売数を維持するストリーミングマテリアライズドビューを作成します。ストリームプロセッサは、sample_supplies.sales コレクションの変更ストリームを読み取り、sample_supplies.sales_by_channel コレクションに書き込み (write)ます。

注意

ストリームプロセッサーは、また Apache Kafka トピックから読み取り、 AWS S3 上の Apache Iceberg テーブルへの書き込み (write) を行うこともできます。詳細については、Apache Kafka Broker および $iceberg 集計ステージ。を参照してください。

前提条件

Stream Processingプロセッサーを作成する前に、ソースデータを保持するクラスターへの Atlas 接続を含む Stream Processingワークスペースが必要です。接続を追加するには、接続の管理を参照してください。

このセクションのコマンドをクラスターに対して実行します。接続するには、「mongosh を使用してクラスターに接続する」を参照してください。

ソースコレクションを準備し、ビューをシードするには、次の手順に従ってください。

サンプルデータセットをロードします。

この手順では、sample_supplies データセットの sales コレクションを使用します。サンプルデータのロード方法を学ぶには、Atlas 配置へのサンプルデータのインポートを参照してください。

販売コレクションにステータスフィールドを追加します。

ストリームプロセッサが完了して返品された販売を検出できるように、status フィールドを追加します。クラスターに対して次のコマンドを実行します。

db.sales.updateMany(   { status: { $exists: false } },   { $set: { status: "completed" } })

{  acknowledged: true,  insertedId: null,  matchedCount: 5000,  modifiedCount: 5000,  upsertedCount: 0}

変更ストリームの事前および書き込みイメージを有効にします。

ストリームプロセッサーが各変更の影響を計算できるように、ソースコレクションでプレイメージと書き込みイメージを有効にします。クラスターに対して次のコマンドを実行します。

db.getSiblingDB("sample_supplies").runCommand({   collMod: "sales",   changeStreamPreAndPostImages: { enabled: true }})

ビューに現在の結果をシードします。

クラスターに対して次のバッチする集計を実行し、ビューに現在のカウントを移入します。

db.sales.aggregate([   { $match: { status: "completed" } },   { $group: { _id: "$purchaseMethod", active_count: { $sum: 1 } } },   { $merge: {      into: "sales_by_channel",      whenMatched: "replace",      whenNotMatched: "insert"   } }])

シードされたカウントを確認するには、ビューをクエリします。sales_by_channel コレクションには、購入方法ごとに 1 つのドキュメントが含まれています。

db.sales_by_channel.find()

[  { _id: 'Phone', active_count: 596 },  { _id: 'Online', active_count: 1585 },  { _id: 'In store', active_count: 2819 }]

このシード集計は、それ自身がオンデマンドマテリアライズドビューです。2 つのビュータイプは相補的です。バッチする集計でオンデマンドマテリアライズドビューを初期化し、同じコレクションを現在の状態に保つストリームプロセッサーを起動できます。オンデマンドビューはストリーミングマテリアライズドビューになります。

注意

パイプラインでウィンドウステージを使用しない場合は、代わりに initialSync を使用してビューにシードすることができます。この場合、ストリームプロセッサーはまず、ソースコレクション内の存在するすべてのドキュメントを挿入イベントとして取り込み、次に新しい変更イベントを処理します。$source オプション initialSync について学ぶには、MongoDB コレクション変更ストリーム。を参照してください。

手順

Atlas UI または mongosh を選択してストリームプロセッサーを作成します。

インターフェース

ビューが最新の状態に保たれていることを確認します

このセクションのコマンドをクラスターに対して実行します。接続するには、「mongosh を使用してクラスターに接続する」を参照してください。

プロセッサーを開始すると、sales コレクションへの変更は数秒以内に sales_by_channel を更新します。これを確認するには、新しい完了したオンライン販売を挿入します。

db.sales.insertOne({   saleDate: new Date(),   purchaseMethod: "Online",   status: "completed",   items: [],   customer: {},   couponUsed: false})

プロセッサーは Online カウントを増分し、ウィンドウ境界を lastWindowStart にレコードします。

db.sales_by_channel.find()

[  { _id: 'Phone', active_count: 596 },  {    _id: 'Online',    active_count: 1586,    lastWindowStart: ISODate('2026-07-23T15:18:01.000Z')  },  { _id: 'In store', active_count: 2819 }]

カスタマーがその販売を返品すると、Online カウントはシード値に戻ります。

db.sales.updateOne(   { purchaseMethod: "Online", status: "completed" },   { $set: { status: "returned" } })

db.sales_by_channel.find()

[  { _id: 'Phone', active_count: 596 },  {    _id: 'Online',    active_count: 1585,    lastWindowStart: ISODate('2026-07-23T15:18:13.000Z')  },  { _id: 'In store', active_count: 2819 }]

例

これらの例では、Atlas Stream Processing の例リポジトリの smv の例について説明します。優先順位ごとに 1 つのドキュメントを保持し、開始されたサポートチケットの実行中のカウントを保持する queue_stats コレクションを維持します。ストリームプロセッサは support_tickets 変更ストリームを読み取り、チケットが開始、解決、削除されるたびに各カウントを更新します。

パイプラインは、チケットが開いたときに +1、チケットが解決または削除されたときに -1 だけチケットの優先順位のカウントを調整します。集計には5つのステージがあります。

$source ステージは、事前および書き込みのイメージを使用して support_tickets 変更ストリームを読み取ります。
$addFields ステージは、operationType 上の $switch を使用して _delta を計算し、_priority をグループキーとして抽出します。
$match ステージは、差分がゼロのイベントを削除します。
$tumblingWindow ステージは、1 秒の各ウィンドウ内で優先順位によってデルタを合計します。
$merge ステージは、再生時の二重カウントを避けるために lastWindowStart をハイウォーターマークとして使用して、各ウィンドウの差分を実行中の合計に追加します。

[  {    "$source": {      "connectionName": "<connection-name>",      "db": "support",      "coll": "support_tickets",      "config": {        "fullDocument": "required",        "fullDocumentBeforeChange": "required"      }    }  },  {    "$addFields": {      "_delta": {        "$switch": {          "branches": [            {              "case": {                "$and": [                  { "$eq": ["$operationType", "insert"] },                  { "$eq": ["$fullDocument.status", "open"] }                ]              },              "then": 1            },            {              "case": {                "$and": [                  { "$eq": ["$operationType", "update"] },                  { "$eq": ["$fullDocumentBeforeChange.status", "open"] },                  { "$eq": ["$fullDocument.status", "resolved"] }                ]              },              "then": -1            },            {              "case": {                "$and": [                  { "$eq": ["$operationType", "delete"] },                  { "$eq": ["$fullDocumentBeforeChange.status", "open"] }                ]              },              "then": -1            }          ],          "default": 0        }      },      "_priority": {        "$ifNull": [          "$fullDocument.priority",          "$fullDocumentBeforeChange.priority"        ]      }    }  },  { "$match": { "_delta": { "$ne": 0 } } },  {    "$tumblingWindow": {      "boundary": "processingTime",      "interval": { "size": 1, "unit": "second" },      "pipeline": [        {          "$group": {            "_id": "$_priority",            "open_count": { "$sum": "$_delta" },            "windowStart": {              "$first": { "$meta": "stream.window.start" }            }          }        }      ]    }  },  {    "$merge": {      "into": {        "connectionName": "<connection-name>",        "db": "support",        "coll": "queue_stats"      },      "whenMatched": [        {          "$set": {            "open_count": {              "$cond": [                { "$gt": ["$$new.windowStart", { "$ifNull": ["$lastWindowStart", { "$toDate": 0 }] }] },                { "$add": ["$open_count", "$$new.open_count"] },                "$open_count"              ]            },            "lastWindowStart": {              "$max": [                { "$ifNull": ["$lastWindowStart", { "$toDate": 0 }] },                "$$new.windowStart"              ]            }          }        }      ],      "whenNotMatched": "insert"    }  }]

queue_stats コレクション内の各ドキュメントは、次のようになります。

{ _id: "P1", open_count: <num>, lastWindowStart: <timestamp> }

1 つのイベントで 2 つのグループキーを調整する必要がある場合は、スカラー _delta を _adjustments 配列に置き換え、配列を調整ごとに 1 つのドキュメントに分散させます。単一キーパイプラインを次のように変更します。

各 $switch ブランチが影響を受けるグループキーごとに 1 つの要素を含む _adjustments 配列を返すように $addFields ステージを置き換えます。エスカレーションは 2 つの要素を返します。
$addFields の後に $unwind ステージと $set ステージを追加します。$unwind ステージは、各イベントを調整ごとに 1 つのドキュメントに分裂し、空の配列を削除します。これにより、$match ステージが置き換えられます。$set ステージは、_adjustments フィールドをトップレベルの _priority と _delta にプロモートし、残りのステージが変更されないまま機能するようにします。

単一キーパイプラインのステージ 2 と 3 を次のように置き換えます。

// Stage 2 (replacement): Compute an _adjustments array.{  $addFields: {    _adjustments: {      $switch: {        branches: [          {            case: { $and: [              { $eq: ["$operationType", "insert"] },              { $eq: ["$fullDocument.status", "open"] }            ]},            then: [{ _priority: "$fullDocument.priority", _delta: 1 }]          },          {            case: { $and: [              { $eq: ["$operationType", "update"] },              { $eq: ["$fullDocumentBeforeChange.status", "open"] },              { $eq: ["$fullDocument.status", "resolved"] }            ]},            then: [{ _priority: "$fullDocumentBeforeChange.priority",              _delta: -1 }]          },          {            case: { $and: [              { $eq: ["$operationType", "update"] },              { $eq: ["$fullDocument.status", "open"] },              { $ne: ["$fullDocument.priority",                "$fullDocumentBeforeChange.priority"] }            ]},            then: [              { _priority: "$fullDocumentBeforeChange.priority",                _delta: -1 },              { _priority: "$fullDocument.priority", _delta:  1 }            ]          },          {            case: { $and: [              { $eq: ["$operationType", "delete"] },              { $eq: ["$fullDocumentBeforeChange.status", "open"] }            ]},            then: [{ _priority: "$fullDocumentBeforeChange.priority",              _delta: -1 }]          }        ],        default: []      }    }  }},
// Stage 3 (replacement): Fan out into one document per adjustment,// then lift the adjustment fields back to the top level.{ $unwind: "$_adjustments" },{  $set: {    _priority: "$_adjustments._priority",    _delta: "$_adjustments._delta"  }}

詳細情報

このガイドのステージと概念についてさらに学ぶには、次のリソースを参照してください。

ストリームプロセッサーの作成、開始、停止、モニターについては、ストリームプロセッサーの開発と管理を参照してください。
Atlas Stream Processing がサポートする集計ステージについて学ぶには、Stream Processing 用集計パイプラインステージを参照してください。
Stream Processing が読み取り可能なソースについて学ぶには、$source ステージ (Stream Processing)。を参照してください。
ウィンドウステージについて学ぶには、Stream Processor Windows. を参照してください。
オンデマンドのマテリアライズドビューの詳細については、「オンデマンドのマテリアライズドビュー」を参照してください。

戻る

階層サイズガイド

マルチコレクションの最初の同期を構成する