このガイドでは、重要なワークフローの具体的な手順を含む、Atlas Stream Processing を管理するためのベストプラクティスについて説明します。
ネットワーキング
Each stream processor relies on connections to data sources and sinks. Atlas Stream Processing encrypts all data in transit using TLS/SSL, but basic connections still pass data through the public internet. To achieve optimal performance and security with Atlas Stream Processing, consider the practices described in this section.
ピアリングとルート テーブル
To ensure secure communication between Atlas Stream Processing and private external systems such as self-managed Apache Kafka clusters or private APIs, use VPC or VNet peering connections. These connections shield your data from exposure to the public internet. Note that in addition to establishing the peering connection, you must explicitly configure route tables for your application's VPC to direct traffic to your Atlas VPC CIDR block. To learn more about how to configure your route table, consult the VPC documentation for your external provider.
Atlas プロジェクト間接続
To route streaming data between different Atlas projects within the same organization, Atlas Stream Processing supports Cross Project Connections. Cross Project Connections transmit data without exposing clusters to the private internet, which ensures privacy without the need to manually configure VPC peering.
Passwordless AWS Authentication
Atlas Stream Processing supports the following AWS integrations:
S3
Kinesis Data Streams
Lambda (as
$externalFunction)
Each of these integrations is compatible with Unified AWS Access. Configuring Unified Access eliminates the need to store static access IDs or Secret credentials in MongoDB, and is the recommended authentication model for all AWS integrations.
Kafka OIDC
Atlas Stream Processing supports authenticating to Apache Kafka brokers using OIDC with JSON Web Tokens. However, this feature currently supports only Identity Providers accessible over public networks. This feature isn't supported for Identity Providers within a customer's VPC.
フォールトトレランス
Atlas Stream Processing provides various mechanisms to ensure reliable, robust service in common failure scenarios, including checkpoints, dead letter queues, and failover processors.
Atlas Stream Processing は、30 秒のハートビートを使用してストリーム プロセッサの健全性を監視します。ストリーム プロセッサが 30 秒を超えてハートビートを送信しない場合、Atlas Stream Processing はプロセッサを自動的に再起動します。
Atlas Stream Processing automatically manages topology changes for sharded integrations like AWS Kinesis.
デッド文字キュー
You can configure Atlas Stream Processing to route unprocessed data to a dead letter queue (DLQ). You can then view unprocessed records in the DLQ to remediate processing failures.
DLQ を調べるときは、次の一般的な処理エラーを探します。
ペイロード解析
受信レコードの形式が正しくないと、解析に失敗します。これは、多くの場合、無効なJSONが原因です。
式評価
処理パイプライン内の式が正しく評価されない。
例
Your pipeline includes the
$emit.config.tombstoneWhenconfiguration field to delete Kafka records and a given document is missing the field to be evaluated.サイズ制限
The outgoing record exceeds the size limit of your target sink, such as
1 MBfor AWS Kinesis.キー生成
プロセッサが Sink 用のパーティションキーを生成する際、エラーが発生しました。
レイテンシデータ
Your processor uses windows and a given document arrived after the configured allowed lateness period expires.
処理失敗の原因を分析した後、 Atlas Stream Processingパイプラインに入力する前にデータを修正するために、上流のデータソースを変更するか、エッジケースを処理するようにパイプラインを変更します。
Atlas Stream Processing 構成の修正をガイドために、DLQ 内のレコードの次の修正オプションを使用することを検討してください。
破棄
レコードが高度に不正であるか、関連性のない場合（時間に依存するシステムで非常に遅延なデータが存在する場合など）は、コレクションからそのドキュメントを削除します。
手動修正
マイナーなJSONエラーを含むレコードや、重要な意味を持つレコードの場合は、フィールドを手動で調整し、ターゲット シンクに直接挿入します。
自動再処理:
回復可能な関連データが大量にある場合は、クリーンされたデータをターゲット シンクにルーティングする前に、DLQ から読み取り、既知の障害条件に変換を適用するセカンダリスクリプトまたは追加のストリーム プロセッサを作成します。
フェイルオーバー
Atlas Stream Processing provides failover processors to prevent service outage in the event of region-wide system failure.
フェイルオーバー構成の原則
次の原則によって、Atlas Stream Processing ワークロードを保護するための堅牢なフェイルオーバー構成を定義できます。
複数のリージョンに Atlas クラスターを配置します。
リージョンフェイルオーバーイベントでストリーム プロセッサを継続的に操作には、ストリーム プロセッサが接続する Atlas クラスターがマルチリージョンクラスターである必要があります。この原則により、プライマリ リージョンがダウンしても、ソース接続またはシンク接続はアクティブのままになります。
各リージョンに対して Atlas Stream Processing ワークスペースを作成します。
マルチリージョンクラスターを配置したら、各リージョンのワークスペースを作成します。この原則により、ストリーム プロセッサとそれをサポートする Atlas データベース間の接続のレイテンシが最小限に抑えられます。
接続レジストリにバックアップ接続を入力します。
Configure connections for each source and sink in each region's workspace. Ensure that you configure connections not only for basic connections, but for VPC and private endpoint connections.
接続固有の障害復旧の構成
ストリーム プロセッサのサービスの継続を確保するには、それが依存するすべての外部プロバイダーもフェイルオーバーを有効にする必要があります。各外部プロバイダーのソースまたは Sink が、ドキュメントに従って適切に構成されていることを確認します。
例
ストリーム プロセッサがKafkaクラスターをソースまたはシンクとして使用する場合は、Atlas Stream Processing がコンシューマー グループ オフセットから処理を再開できるように、コンストラクター オフセットを保存するようにクラスターを構成します。
ルーチン ドライバーの実行を実行します。
フェイルオーバーのドライバーを定期的に実行して、フェイルオーバー構成を検証してください。これらのドライバーの実行は、構成がネットワークに接続され、チェックポイントを正しくロードされるようにするだけでなく、フェイルオーバーのワークフローを訓練して、必要な場合に実行しやすくする両方の目的で使用されます。フェイルオーバーのドライバー実行の定期的なケイデンスを確立することを検討し、フェイルオーバー構成を変更するたびにドライバーを実行してください。
ストリーム プロセッサのフェイルオーバーを設定する
After ensuring best practices are in place, create a stream processor and verify your failover configuration.
Atlas Stream Processing は、次の構成を持つプロセッサのみの自動リージョン フェイルフェイルオーバーをサポートします。
Atlas ソースとシンク
Atlas source and Apache Kafka sink
For more information about automated failover, see Failover Processor.
ただし、ソース接続とシンク接続の任意の組み合わせで、手動フェイルオーバーをサポートする Atlas Stream Processing アーキテクチャを構成できます。
強制フェイルオーバーの開始
強制フェイルオーバーは、ワークスペース レベルまたはストリーム プロセッサ レベルで開始できます。
To initiate stream processing workspace failover, follow the procedure described in Trigger stream processing workspace Failover.
To initiate individual stream processor failover, follow the procedure described in Initiate Failover for One Stream Processor.
フェイルオーバー イベントへの応答
リージョン サービス停止時イベントが発生した場合 :
サービスの停止時を確認します。
When you notice symptoms of a service outage, review your preferred status report to verify the event. You can view this information on the MongoDB Cloud Status page or your cloud provider's status page. You can configure Atlas Stream Processing Alerts. Note the timestamp of the failure event.
フェイルオーバープロセッサを起動します。
Follow the procedure described in Initiate Failover for One Stream Processor. To ensure service resumes from the last consumed stream event, pass the timestamp you noted previously as the startAtOperationTime parameter.
Processors using Apache Kafka sources resume according to consumer group offsets.
サービスが再開されたことを確認します。
After completing failover, periodically check the status of your workspace's primary region. You can view this information on the MongoDB Cloud Status page or your cloud provider's status page. You can also configure Atlas Stream Processing Alerts. If you use automatic primary processor suppression, you can configure the script to return notifications when it detects primary processors automatically restarting.
プライマリ プロセッサを再起動します。
Restart the stream processors in your primary region. To ensure service resumes from the last consumed stream event, pass the timestamp you noted previously as the startAtOperationTime parameter.
Processors using Apache Kafka sources resume according to consumer group offsets.
フェイルオーバーの動作
リージョン フェイルフェイルオーバーまたは 手動フェイルオーバーをトリガーするようにシステムを構成する場合は、次の点を考慮してください。
When regional failover occurs, Atlas Stream Processing promotes the inactive failove processors in your failover region to active processors. These processors resume stream processing from the last available checkpoint.
Atlas Stream Processing チェックポイントの古いバージョンは 10 分以下であることが保証されます。ただし、フェイルオーバープロセッサは、開始時の最後のチェックポイント以降にプライマリ プロセッサが処理したデータを再処理します。フェイルオーバーは、少なくとも 1 回のプロセシング セマンティクスを保証します。
リージョン フェイルフェイルオーバーイベント中にフェイルオーバープロセッサの起動に失敗した場合でも、アクティブなプロセッサのロールを引き受けます。その後、基礎となる原因のトラブルシューティングを行い、フェイルオーバーリージョン で通常どおりプロセッサを再起動できます。
観察可能性
Atlas Stream Processing provides various observability tools to assess the performance and health of your workspaces, connections, and processors. Additionally, to track regional service status and outages, you can visit the MongoDB Cloud Status page.
Atlas Stream Processing のスケーリング
Atlas Stream Processing provisions resources per stream processor. To learn more, see the Tier Selection Guide.
For Atlas Stream Processing processors that operate on Atlas clusters, ensure you also scale your both your processors and clusters proportionate to your pipeline I/O needs. In particular, evaluate processors that perform a high volume of $lookup or $merge operations as candidates for cluster upscaling.
変更ストリームに関する考慮事項
Atlas Stream Processing can use a Atlas cluster $source to consume change stream events. To ensure optimal performance and reliability for such stream processors, review the following considerations:
oplog window サイズ
Stream processors leverage the
$sourcecluster's oplog to resume service after stoppage. If a processor is down for a period greater than the oplog retention period, the resume token can point to truncated data, causing aChangeStreamHistoryLosterror.oplog のサイズ設定をストリーム処理システムの許容ダウンタイムと整合させることで、このリスクを軽減します。
再開オプション
Atlas Stream Processing supports resuming change-stream
$source``s from either the last stored checkpoint with ``resumeFromCheckpointor a specific time withstartAtOperationTime. To avoid gaps or duplications in processed data, define a standardized policy around which resumption options you use in a given scenario.バックログ取り込み
大規模な名前空間で動作するストリーム プロセッサでは、キャッチアップ処理の初期期間が長く発生する可能性があります。この初期キャッチアップ期間を短縮するには、ストリーム プロセッサの名前空間の範囲を減らすか、より新しい開始点を選択することを検討してください。
変更ストリームのイメージ
Auditing, compensating transactions, and undo operations require access to pre-processing document state. When you work with Atlas as a source, enable Document Pre- and Post-Images to support these operations. When you roll out support for document pre- or post-images, verify coverage and stability across your system.
``$replaceRoot`` ステージ
Change stream sources return documents that bundle operational metadata with the content of the affected document. Writing these documents to a sink requires consumers to work with the change-event schema rather than the business logic of the source document. Unless you require the metadata, use the
$replaceRootstage to promote the source content subdocument before further processing.結合指向パイプライン
Atlas Stream Processing supports the
$lookupstage to enable join-like enrichments of streaming data. These operations can increase processing latency. Consider limiting your Atlas Stream Processing pipelines to only those$lookupoperations that must be performed before writing to a sink, reserving more complex enrichments to precomputed views.古いチェックポイント
チェックポイントにはイベント履歴のみが含まれ、処理ロジックは含まれません。チェックポイントから操作を再開すると、ストリーム プロセッサは現在のパイプラインのロジックを古いイベントに適用します。これにより、既存の下流の状態と新しい出力が不一致になる可能性があります。大幅に異なる処理セマンティクスが必要な場合は、チェックポイントから再開するのではなく、新しいストリーム プロセッサを導入することを検討してください。
プランニングを再起動する
再起動中、ストリーム プロセッサでは新しいイベントのスループットが低下する可能性があります。システムの回復力を高めるには、 一時的に減少したキャパシティーを考慮して再起動をスケジュールします。
フェイルオーバー プランニング
リージョン間でストリーム プロセッサ チェックポイントを共有し、安全な再開ポイントの選択、出力の整合性を検証するための標準手順を開発し、リージョンフェイルオーバーイベントがデータ品質の問題につながることがなくなります。