複数コレクションの最初の同期の設定

マルチコレクション最初の同期の特権

A multi-collection initialSync operation has the following traits:

• Syncs collections in list order. With coll: ["a", "b", "c"], Atlas Stream Processing drains a's partitions before b's, and b's before c's, except where spillover fills idle capacity with partitions from a later collection.

• 完了したコレクションを再同期せずに再開します。 Atlas Stream Processing はどのコレクションが終了したかを追跡するため、再起動されたストリーム プロセッサは完了したコレクションをスキップし、進行中のコレクションを最後に同期されたドキュメントから再開します。すでに出力したドキュメントが再配信される可能性があるため、繰り返されるドキュメントを冪等で処理するようにシンクを設計します。

• コレクションレベルの詳細な同期の進行状況を公開します。 stats().stats.operatorStats[0].targetStats reports sync progress and document counts for each target collection. To learn more, see Verify Initial Sync Progress.

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

Multi-collection initialSync differs from single-collection sync in ways that affect how you configure and operate your stream processor:

• Parallelism is global, not per collection. Setting initialSync.parallelism caps the number of concurrent partition reads across the entire set of target collections, not per collection.

• キャッチアップ スコープは、ソース スコープの後にします。同期フェーズの後、Atlas Stream Processing は、ソースのスコープに一致する変更ストリームを開きます。たとえば、データベース全体のソースの場合はデータベースレベルの変更ストリーム、クラスター全体のソースの場合はクラスターレベルの変更ストリーム、またはフィルタリングされた変更ストリームはソースのスコープに一致します。コレクションの明示的なリスト。これを個別に構成する必要はありません。

• コレクションの同期に失敗すると、プロセッサ全体が失敗します。単一コレクション同期の場合と同様に、Atlas Stream Processing がリスト内のいずれかのコレクションの同期に失敗すると、ストリーム プロセッサ全体が失敗します。

• 一部のソース側構成は宛先に複製されません。 initialSync では、インデックス、Cappedコレクション、時系列コレクション、バリデーター、デフォルトの照合、ビュー定義などのコレクションオプションは複製されません。ストリーム プロセッサを起動する前に、宛先でこれらを作成します。 Atlas Stream Processing は、ストリーム プロセッサの起動後に作成したコレクションも検出しません。同期するコレクションのセットは、その点で 固定されています 。

前提条件

この手順を完了する前に、以下の手順を行ってください。

• A stream processing workspace with Atlas connections to both the source and destination clusters. To add a connection, see Manage Connections.

• A processor tier that supports the number of collections you plan to list. To learn more, see Resource Allocation.

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

ソース コレクションを準備するには、次の手順を実行します。

db.getSiblingDB("sample_analytics").runCommand({
   collMod: "customers",
   changeStreamPreAndPostImages: { enabled: true }
})

{ ok: 1, ... }

db.getSiblingDB("sample_analytics").runCommand({
   collMod: "accounts",
   changeStreamPreAndPostImages: { enabled: true }
})

{ ok: 1, ... }

手順

Select the Atlas UI or mongosh to configure the stream processor.

最初の同期の進行状況を確認する

After you start the stream processor, call sp.<processor-name>.stats() with the verbose option to view sync progress for each target collection. The source operator's entry in stats.operatorStats reports a targetStats array with one element per collection:

sp.replicate_analytics_sp.stats(
  { options: { verbose: true } }
).stats.operatorStats[0].targetStats

[
  {
    db: 'sample_analytics',
    coll: 'customers',
    inputMessageCount: Long('500'),
    inputMessageSize: 672326,
    initialSync: {
      estimatedDocs: Long('500'),
      copiedDocs: Long('251'),
      status: 'in_progress'
    }
  },
  {
    db: 'sample_analytics',
    coll: 'accounts',
    inputMessageCount: Long('1746'),
    inputMessageSize: 1883681,
    initialSync: {
      estimatedDocs: Long('1746'),
      copiedDocs: Long('1746'),
      status: 'completed'
    }
  }
]

Each target's initialSync.status field cycles through pending, in_progress, and completed as Atlas Stream Processing syncs that collection, or reports failed if the sync fails. estimatedDocs and copiedDocs report the estimated and completed document counts for that one collection.

To view total progress across every target collection, use the source operator's own initialSync field, which sums estimatedDocs and copiedDocs across all collections:

sp.replicate_analytics_sp.stats(
  { options: { verbose: true } }
).stats.operatorStats[0].initialSync

{ estimatedDocs: Long('2246'), copiedDocs: Long('1997') }