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

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

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

前提条件

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

• ソースクラスターと宛先クラスターの両方に Atlas が接続されているストリーム処理ワークスペース。接続を追加するには、「 接続の管理 」を参照してください。

• 一覧表示するコレクションの数をサポートするプロセッサ階層。詳しくは、「 リソース割り当て 」を参照してください。

このセクションのコマンドをクラスターに対して実行します。接続するには、「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, ... }

手順

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

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

ストリーム プロセッサを起動した後、verbose オプションを指定して sp.<processor-name>.stats() を呼び出し、各ターゲットコレクションの同期進行状況を表示します。 stats.operatorStats のソース演算子のエントリでは、コレクションごとに 1 つの要素を含む targetStats 配列が報告されています。

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

各ターゲットの 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

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