initialSync Atlas Stream Processingストリームプロセッサが、変更ストリームのテーリングを開始する前に、Atlasコレクション内の既存のドキュメントを挿入changeEventドキュメントであるかのように取り込めるようにします。単一の $source ステージで、一度に複数のコレクションに対して initialSync を実行できるようになりました。
このページでは、単一コレクション同期と複数コレクション同期の特性を比較し、複数コレクション同期の設定手順を説明し、これを基盤として構築されたクロスクラスターレプリケーションパイプラインをデモンストします。
単一コレクションの最初の同期との比較
$source ステージの coll フィールドは、単一のコレクション名またはコレクション名の配列のいずれかを受け入れます。配列内で名前を指定するすべてのコレクションは、db フィールドで指定するデータベースに属している必要があります。単一の $source ステージでは、複数のデータベースからのコレクションを同期することはできません。提供する値によって同期スコープが決まります：
スコープ | $source 構成 | 動作 |
|---|---|---|
単一のコレクション |
| Atlas Stream Processing は、名前付きコレクションを同期します。 |
Explicit list |
| Atlas Stream Processing は、リスト内の各コレクションを、リストされた順序で同期します。 |
マルチコレクションの最初の同期の特性
マルチコレクションの initialSync 操作には、次の特性があります。
リスト順にコレクションを同期します。
coll: ["a", "b", "c"]の場合、Atlas Stream Processing はaのパーティションをbより先に、bのパーティションをcより先にドレインします。ただし、スピルオーバーによって、後のコレクションのパーティションでアイドル状態の容量が使用される場合を除きます。完了したコレクションを再同期せずに再開します。Atlas Stream Processingは、どのコレクションが完了したかを追跡するため、再起動されたストリームプロセッサは完了したコレクションをスキップし、進行中のコレクションを最後に同期されたドキュメントから再開します。すでに送信されたドキュメントを再送信する可能性があるため、シンクが重複したドキュメントを取り扱うことができるように設計してください。
コレクションレベルでの細かい同期の進捗状況を表示します。
stats().stats.operatorStats[0].targetStats各ターゲットコレクションの同期の進行状況とドキュメント数を報告します。詳細については、「最初の同期の確認」を参照してください。
考慮する必要がある動作の違い
複数のコレクション initialSync は単一のコレクション同期とは異なります。これにより、ストリーム プロセッサーの構成と操作方法に影響します：
平行処理はグローバルであり、コレクションごとではありません。
initialSync.parallelismを設定すると、ターゲット コレクションのセット全体にわたって同時実行されるパーティション読み取りの数を制限します。コレクションごとではありません。キャッチアップスコープはソーススコープに従います。同期フェーズの後、Atlas Stream Processingはソースのスコープに一致する変更ストリームを開きます。データベースレベルの変更ストリーム（データベース全体がソースの場合）、クラスターレベルの変更ストリーム（クラスター全体がソースの場合）、または明示的なコレクションリストに対するフィルターされた変更ストリームです。これを個別に設定する必要はありません。
コレクション同期失敗は、プロセッサー全体を失敗させます。単一コレクション同期と同様に、Atlas Stream Processing がリスト内のいずれかのコレクションの同期に失敗した場合、ストリーム プロセッサー全体が失敗します。
一部のソース側の構成は、宛先にレプリケートされません。
initialSyncインデックス、上限付きコレクションなどのコレクションオプション、時系列コレクション、バリデーター、デフォルト照合、またはビュー定義はレプリケートされません。ストリーム プロセッサーを開始する前に、宛先でこれらを作成します。Atlas Stream Processing も、ストリーム プロセッサーの開始後に作成したコレクションを検出しません。同期するコレクションのセットは、その時点で固定されます。
マルチコレクションの最初の同期を構成する
次の手順では、sample_analytics データセットの 2 つのコレクション、customers（顧客プロファイルを保持）と accounts（各顧客の金融口座を保持）を使用します。この手順では、ある Atlas クラスターから別の Atlas クラスターに両方のコレクションをコピーし、ドキュメントを最新の状態に維持するストリームプロセッサを設定します。
前提条件
この手順を完了するには、以下が必要です。
Atlas 接続を使用して、ソース クラスターとデスティネーション クラスターの両方に接続する Stream Processing ワークスペース。接続を追加するには、接続の管理を参照してください。
リストする予定のコレクションの数をサポートするプロセッサ階層。詳しく学ぶには、「リソース割り当て」を参照してください。
このセクションのコマンドをクラスターに対して実行します。接続するには、「mongosh を使用してクラスターに接続する」を参照してください。
次の手順に従って、ソース コレクションを準備します。
サンプル データセットをロードします。
この手順では、sample_analytics データセットの customers コレクションと accounts コレクションを使用します。サンプル データのロード方法を学ぶには、Atlas 配置へのサンプル データのインポートを参照してください。
変更ストリームの事前および書き込みイメージを有効にします。
customers コレクションと accounts コレクションで事前イメージと書き込みイメージを有効にします。これは、このプロシージャの $source ステージで fullDocument: "required" が設定されるためです。クラスターに対して次のコマンドを実行します。
db.getSiblingDB("sample_analytics").runCommand({ collMod: "customers", changeStreamPreAndPostImages: { enabled: true }})db.getSiblingDB("sample_analytics").runCommand({ collMod: "accounts", changeStreamPreAndPostImages: { enabled: true }})fullDocument: "required" の詳細については、「MongoDB コレクションの変更ストリーム」を参照してください。
手順
Atlas UIまたはmongoshを選択して、ストリームプロセッサを構成します。
最初の同期の進捗状況を確認する
stream processor を起動した後、verbose オプションを指定して sp.<processor-name>.stats() を呼び出し、各ターゲット コレクションの同期の進行状況を表示します。stats.operatorStats のソース演算子のエントリには、コレクションごとに 1 つの要素を持つ `targetStats` 配列が報告されています：
sp.replicate_analytics_sp.stats( { options: { verbose: true } }).stats.operatorStats[0].targetStats各ターゲットのinitialSync.statusフィールドは、Atlas Stream Processing がそのコレクションを同期すると、pending、in_progress、completedの順に変化します。同期が失敗した場合はfailedを報告します。estimatedDocs と copiedDocs は、その1つのコレクションについて、見積りと完了したドキュメント数を報告します。
すべてのターゲット コレクションの全体の進捗状況を表示するには、ソース演算子自身の initialSync フィールドを使用します。このフィールドは、すべてのコレクションにおける estimatedDocs と copiedDocs の合計を計算します。
sp.replicate_analytics_sp.stats( { options: { verbose: true } }).stats.operatorStats[0].initialSync例
前述の手順では、Atlas クラスター間で 2 つのコレクションをレプリケートします。次の例では、同じパイプラインを別のコレクションのセットに適用します。$source ステージの db と coll の値のみが変更されます。
詳細情報
このガイドのステージと概念についてさらに学ぶには、次のリソースを参照してください。
$sourceステージとそのinitialSyncフィールドについて学ぶには、「MongoDB コレクションの変更ストリーム」を参照してください。プロセッサの階層が同期可能なコレクションの数にどのように影響するかについては、「リソースの割り当て」を参照してください。
Atlas Stream Processing が進行中の同期にチェックポイントを設定する方法について詳しくは、チェックポイントを参照してください。
ストリーム プロセッサーの作成、開始、停止、モニターについては、ストリーム プロセッサーの開発と管理を参照してください。