Atlas Stream Processingの操作

ピアリングとルート テーブル

Atlas Stream Processingと、自己管理型のApache KafkaクラスターやプライベートAPIなどのプライベートな外部システムとの間で安全な通信を確保するには、VPCまたはVNetピアリング接続を使用します。これらの接続により、データがパブリックインターネットにさらされるのを防ぐことができます。ピアリング接続の確立に加えて、アプリケーションのVPCのルートテーブルを明示的に構成し、トラフィックをAtlas VPC CIDRブロックにルーティングする必要があることに注意してください。ルートテーブルの構成方法の詳細については、外部プロバイダーのVPCドキュメントを参照してください。

Atlas プロジェクト間接続

同じ組織内の異なる Atlas プロジェクト間でストリーミング データをルーティングするには、Atlas Stream Processing は Cross Project Connections をサポートしています。Cross プロジェクト Connections は、クラスターをプライベートインターネットに公開することなくデータを送信し、VPC ピアリングを手動で構成する必要なくプライバシーを確保します。

パスワードレスAWS認証

Atlas Stream Processing は、次の AWS 統合をサポートしています。

• S3

• Kinesis Data Streams

• Lambda ($externalFunction として)

これらの統合はそれぞれ、Unified AWS Access と互換性があります。統合アクセスを設定すると、MongoDB に静的アクセス ID やシークレット認証情報を保存する必要がなくなり、すべての Amazon Web Services 統合における推奨の認証モデルとなります。

Kafka OIDC

Atlas Stream Processingは、 OIDC と JSON Web Tokens を使用して、 Apache Kafka ブローカーへの認証をサポートしています。ただし、この機能は現在、パブリックネットワーク経由で到達できるIdPのみをサポートしています。この機能は、カスタマーのVPC内のIdPではサポートされていません。

デッドレターキュー (DLQ)

Atlas Stream Processingを構成して、未処理のデータをデッドレターキュー (DLQ)にルーティングできます。その後、DLQ内の未処理レコードを表示して、プロセシングの失敗を修正できます。

DLQ はドキュメントごとの処理失敗のみをキャプチャします。その他のエラーがある場合はプロセッサの実行中が妨げられ、DLQ レコードは生成されません。このようなエラーには、無効なパイプライン構文または構成と、Atlas Stream Processing システム内のエラーが含まれます。これらのエラーに対して、Atlas Stream Processing はエラー処理と再試行ポリシーを適用します。

ドキュメントの処理に失敗し、ストリーム プロセッサにDLQが構成されていない場合、Atlas Stream Processing はドキュメントを削除します。 などのDLQstats.dlqMessageCount sp.processor.sample()統計では、引き続き失敗がレコード。 でプロセッサをサンプルと、失敗したドキュメントは引き続きサンプル結果に表示されます。

一部のパイプラインステージは、失敗したドキュメントがDLQに到達するかどうかを決定する独自のエラー処理動作を定義します。

DLQを調査する際は、以下のような一般的なプロセシングエラーがないか確認してください。

• ペイロードの解析

  受信したレコードの形式が正しくないため、解析に失敗しました。これは多くの場合、無効なJSONが原因です。

• 式評価

  プロセシングパイプライン内の式が正しく評価されません。

  例

  あなたのパイプラインには、Kafka レコードを削除するための $emit.config.tombstoneWhen 設定フィールドが含まれています。そして、所定のドキュメントには評価するフィールドがありません。

• サイズ制限

  出力レコードが、ターゲット シンクのサイズ制限（例えば、Amazon Web Services Kinesis の 1 MB など）を超えています。

• キー生成

  シンクのパーティションキーを生成する際に、プロセッサでエラーが発生しました。

• 遅延データ

  プロセッサーは ウィンドウ を使用します。そして、設定された 許容された遅延期間が終了した後に指定されたドキュメントが到着しました。

各DLQレコードには、失敗したドキュメントと失敗の詳細が含まれています。完全なDLQドキュメントスキーマについては、 「監視ガイド」を参照してください。

プロセシングの失敗の原因を分析するした後、アップストリームのデータソースを修正してAtlas Stream Processingパイプラインに入る前にデータを修正するか、エッジケースを取り扱うするようにパイプラインを修正します。

Atlas Stream Processingの設定を修正するためのガイドとしてDLQ内のレコードを使用した後は、以下の修正オプションを検討してください。

• 破棄

  レコードに大幅な形式不備があるか、または関連性がない場合（例えば、時間に対して高度な敏感性を持つシステムで大幅に遅れているデータなど）は、コレクションから削除します。

• 手動修正

  軽微なJSONエラーがあるレコードや、ミッションクリティカルなレコードの場合は、フィールドを手動で調整し、ターゲットシンクに直接挿入します。

• 自動再処理:

  復旧可能で関連性の高いデータが大量にある場合は、DLQから読み取り、既知の障害条件に対して変換を適用してから、クリーンアップされたデータをターゲットシンクにルーティングするセカンダリスクリプトまたは追加のストリームプロセッサを作成します。

フェイルオーバー構成の原則

以下の原則は、Atlas Stream Processing ワークロードを保護するための強固なフェイルオーバー構成を定義します。

• Atlas クラスターを複数のリージョンに配置する。

  リージョンフェイルオーバーイベントでストリームプロセッサーの継続的な操作を確保するには、ストリームプロセッサーが接続するすべての Atlas クラスターにマルチリージョンクラスターを使用します。Atlas Stream Processing は単一リージョン クラスターのフェイルオーバーをブロックしません。しかし、この原則により、プライマリ リージョンが停止している場合でも、ソースまたはシンク接続はアクティブな状態を維持します。

• Atlas Stream Processing ワークスペースをリージョンごとに作成します。

  マルチリージョンクラスターを配置後、各リージョンに対してワークスペースを作成します。この原則により、ストリーム プロセッサとそれらをサポートする Atlas データベース間の接続のレイテンシが最小限に抑えられます。

• 接続レジストリにバックアップ接続を追加します。

  各リージョンのワークスペースで、各ソースとシンクの接続を構成します。基本的な接続だけでなく、VPCやプライベートエンドポイントの接続も設定してください。

• 接続固有の障害復旧を設定する

  ストリームプロセッサーのサービス継続性を確保するには、ストリームプロセッサーが依存する外部プロバイダーもフェイルオーバーを有効にする必要があります。各外部プロバイダーのソースまたはシンクが、そのドキュメントに従って正しく構成されていることを確保します。

  例

  ストリームプロセッサがKafkaクラスターをソースまたはシンクとして使用している場合は、コンシューマーオフセットの保持用にクラスターを構成して、Atlas Stream Processingがコンシューマーグループのオフセットから処理を再開できるようにします。

• 定期的にドライランを実行してください。

  フェイルオーバーのドライランを定期的に実行して、フェイルオーバー構成を検証します。これらのドライランは、構成がネットワークに接続され、チェックポイントが正しく読み込まれることを確認するだけでなく、必要に応じて実行しやすくするためにフェイルオーバーワークフローをトレーニングする役割も果たします。フェイルオーバーのドライランを定期的に実行することを検討し、フェイルオーバー構成を変更するたびにドライランを実行してください。

Stream Processorのフェイルオーバーの設定

ベストプラクティスが適用されていることを確認した後、 ストリームプロセッサーを作成してください、そしてフェイルオーバー構成を確認してください。

Atlas Stream Processing は、次の構成を持つプロセッサに対してのみ、自動リージョン間フェイルオーバーをサポートしています。

• Atlas ソースとシンク

• Atlas ソースと Apache Kafka シンク

自動フェイルオーバーの詳細については、「フェイルオーバー Processor」を参照してください。

ただし、ソースとシンクのあらゆる組み合わせに対応してマニュアル フェイルオーバーをサポートする Atlas Stream Processing アーキテクチャを構成できます。

強制フェイルオーバーの開始

ワークスペースレベルまたはストリームプロセッサーレベルのいずれかで、強制フェイルオーバーを開始できます。

Stream Processing ワークスペースのフェイルオーバーを開始するには、Stream Processing ワークスペースのフェイルオーバーをトリガーするに記述されている手順に従ってください。

個別のストリーム プロセッサーのフェイルオーバーを開始するには、1 つのストリーム プロセッサーのフェイルオーバーを開始するに記述されている手順に従ってください。

フェイルオーバー イベントへの対応

リージョナルサービス停止時イベントが発生した場合：

フェイルオーバー動作

リージョンフェイルオーバーまたはマニュアルフェイルオーバーをトリガーするためにシステムを構成する際は、次の点を考慮してください。

• リージョンフェイルオーバーが発生すると、Atlas Stream Processing はフェイルオーバーリージョンの非アクティブなフェイルオーバープロセッサーをアクティブなプロセッサーに昇格させます。これらのプロセッサーは、最後に利用可能な チェックポイント から Stream Processing を再開します。

• Atlas Stream Processingのチェックポイントは、最大10分前までのものが保証されます。ただし、フェイルオーバープロセッサは、開始時に最後のチェックポイント以降にプライマリプロセッサが処理したデータを再処理します。フェイルオーバーは、少なくとも1回のプロセシングセマンティクスを保証します。

• リージョンフェイルオーバーイベント中にフェイルオーバープロセッサの起動が失敗した場合でも、そのプロセッサはアクティブプロセッサのロールを引き継ぎます。その後、フェイルオーバーリージョンで基礎の原因をトラブルシューティングし、通常どおりプロセッサを再起動できます。