Atlas Stream Processing の概要
Atlas Stream Processing を使用すると、Atlas データベースで使用されるのと同じクエリ APIを使用して、複雑なデータのストリームを処理できます。 Atlas Stream Processing を使用すると、以下を行うことができます。
集計パイプラインをビルドし、バッチ処理に固有の遅延なしにストリーミング データで継続的に動作します。
継続的なスキーマ検証を実行して、メッセージが適切に作成されているかどうか、メッセージの破損を検出、遅延データの検出を行います。
Atlas コレクションまたは Apache Kafka に結果を継続的に公開する クラスター、最新のビューとデータの分析を保証します。
Atlas Stream Processing コンポーネントは Atlas プロジェクトに直接属し、Atlas クラスターとは独立して動作します。
データのストリーミング
ストリームは 、1 つ以上のソースから発生するデータの連続したフローであり、追加専用ログの形式をとります。 データストリームの例には、センサーからの温度や負荷の読み取り、金融トランザクションの記録、または変更データをキャプチャするイベントが含まれます。
データ ストリームは、 Apache Kafka トピック または 変更ストリームなどの ソース に基づきます 。その後、処理されたデータを Apache Kafka トピック または Atlas コレクションなどの シンク に書き込むことができます。
データのストリームは、状態が急速に変化するシステムから始まります。 Atlas Stream Processing は、保存中のデータベースの時間や計算上の制約なしに連続データを操作するネイティブ ストリーム プロセシング機能を提供します。
アーキテクチャ
Atlas Stream Processing の主要抽象化はストリーム プロセッサです。 ストリーム プロセッサは、指定されたソースからのストリーミング データに対して継続的に動作し、出力を Sink に書込む MongoDB集計パイプラインクエリです。 詳細については、「 ストリーム プロセッサの構造 」を参照してください。
ストリーム処理はストリーム処理インスタンスで行われます。 各ストリーム プロセシング インスタンスは、以下を関連付ける Atlas 名前空間です。
1 つ以上のワーカー。ストリーム プロセッサを実行するために必要な RAM と CPU を提供します。
クラウドプロバイダーとクラウド リージョン。
ストリーミング データの利用可能なソースとシンクのリストを保存する接続レジストリ。
ユーザー認証を定義するセキュリティ コンテキスト。
ストリーム処理インスタンス自体への 文字列接続。
ストリーム プロセッサを定義すると、それが定義されているストリーム プロセシング インスタンスでのみ使用可能になります。 各ワーカーは最大 4 つの実行中のストリーム プロセッサをホストできます。 Atlas Stream Processing は、ストリーム プロセッサを起動すると、必要に応じて ワーカー をプロビジョニングすることで、ストリーム プロセシング インスタンスを自動的にスケールアップします。 すべてのストリーム プロセッサを停止することで、ワーカーのプロビジョニングを解除できます。 Atlas Stream Processing では、新しいワーカーをプロビジョニングするよりも、既存のワーカーにストリーム プロセッサを割り当てることを常に優先します。
例
proc01からproc08という名前の 8 つのストリーム プロセッサを実行しているストリーム プロセシング インスタンスがあるとします。 proc01からproc04は 1 つのワーカーで実行され、 proc05からproc08は 2 番目のワーカーで実行されます。 proc09という名前の新しいストリーム プロセッサを起動します。 Atlas Stream Processing は、 proc09をホストするための 3 つ目のワーカーをプロビジョニングします。
その後、最初のワーカーでproc03を停止します。 proc09を停止して再起動すると、Atlas Stream Processing はproc09を最初のワーカーに再割り当てし、3 番目のワーカーのプロビジョニングを解除します。
proc09を停止して再起動する前に、 proc10という名前の新しいストリーム プロセッサを起動すると、Atlas Stream Processing は、以前にproc03に割り当てられていたスロットの最初のワーカーにproc10を割り当てます。
スケーリング時に、Atlas Stream Processing では現在実行中のストリーム プロセッサの数のみが考慮されます。実行されていない定義済みストリーム プロセッサはカウントされません。 ストリーム プロセシング インスタンスの階層によって、ワーカーの RAM と CPU の割り当てが決まります。
接続レジストリには 1 つ以上の接続が保存されます。 各接続は、ストリーム プロセッサが外部サービスと交流できるようにするネットワークとセキュリティの詳細の組み合わせに名前を割り当てます。 接続では、次の動作が見られます。
特定のストリーム プロセシング インスタンスの接続レジストリで定義された接続のみが、そのストリーム プロセシング インスタンスでホストされているストリーム プロセッサを処理できます。
各接続は任意の数のストリーム プロセッサを処理できます
単一の接続のみが、特定のストリーム プロセッサのソースとして機能できます。
1 つの接続のみが、特定のストリーム プロセッサのシンクとして機能できます。
接続は、ソースまたはシンクのいずれかとして最初に定義されることはありません。 ストリーム プロセッサがその接続を呼び出す方法に応じて、任意の接続がどちらの機能も提供できます。
Atlas Stream Processing は、マルチテナント インフラストラクチャ上の専用のカスタマー コンテナでストリーム プロセシング ワーカーを実行します。 MongoDB のセキュリティとコンプライアンスの詳細については、 MongoDB トラスト センター を参照してください。
ストリームプロセッサの構造
ストリーム プロセッサは 集計パイプライン の形式をとります。 各プロセッサは$sourceステージで始まり、ソースに接続し、ドキュメント形式でデータの連続ストリームの受信を開始します。 これらのドキュメントは有効なjsonまたはejsonである必要があります。 $sourceの後の各集計ステージは、ストリームからの各レコードを順に消費し、次の 3 つのタイプにグループ化できます。
検証:
$validateステージでは、取り込まれたドキュメントに対してスキーマ検証を実行して、正しく形式化されたドキュメントのみがさらに処理を続行するようにし、誤って形式化されたドキュメントに何が起こるかを判断できます。 検証は任意です。ステートレス操作: 受信データストリームに対して直接実行できる集計ステージまたは演算子。 これらの集計は各ドキュメントを消費、変換、順に渡し、
$sourceと$ mergeステージのいずれかの間の任意の点に表示される可能性があります。ステートフル操作: 制限されたドキュメント セットに対してのみ動作できる集計ステージまたは演算子。これらの集計は、ドキュメント セット全体を一度に消費、変換、および渡すもので、ウィンドウ内にのみ表示されます。
ウィンドウ はストリーミング データを消費し、それを時間区切りのセットに分割するパイプライン ステージです。これにより、 $groupや$avg などの無限データには適用できないステージや演算子を適用できます。 各ストリーム プロセッサには、ウィンドウ ステージが 1 つだけ含めることができます。
取り込まれたデータを処理した後、ストリーム プロセッサは、 $emitステージを使用するストリーミング データ プラットフォーム、または$mergeステージを持つ Atlas データベースにそれを書き込みます。 これらのステージは互いに排他的であり、ストリーム プロセッサにはこのようなステージが 1 つだけ含まれます。
チェックポイント
Atlas Stream Processing は、 チェックポイント ドキュメントを使用してストリーム プロセッサの状態を取得します。 これらのドキュメントは一意の ID を持ち、ストリーム プロセッサ ロジックのフローの対象となります。 ストリーム プロセッサの最後の演算子がチェックポイント ドキュメントの操作を終了すると、Atlas Stream Processing はチェックポイントをコミットし、2 種類のレコードを生成します。
チェックポイントID とそれが属するストリーム プロセッサを検証する単一のレコード
Atlas Stream Processing がチェックポイントをコミットした時点での、関連するストリーム プロセッサの各ステートメント操作の状態を説明するレコードのセット。
中断後にストリーム プロセッサを再起動すると、Atlas Stream Processing は最後にコミットされたチェックポイントをクエリし、記述された状態から操作を再開します。
デッド文字キュー
Atlas Stream Processing は、Atlas データベース コレクションをデッド レター キュー（DLQ）として使用することをサポートしています。 Atlas Stream Processing がデータ ストリームからのドキュメントを処理できない場合、処理失敗の詳細とともにドキュメントの内容をDLQ に書込みます。 ストリーム プロセッサの定義でコレクションをDLQとして割り当てることができます。
詳細については、「ストリーム プロセッサの作成 」を参照してください。
Atlas Stream Processing リージョン
Atlas Stream Processingを使用すると、次の Amazon Web Servicesリージョンでインスタンスをストリーミングできます。
Atlas Stream Processing リージョン | AWS リージョン |
|---|---|
Virginia, USA | us-east-1 |
米国ワシントン州 | us-west-2 |
カナダ（中央） | ca-central-1 |
サンパウロ（ブラジル） | sa-east-1 |
アイルランド | eu-west-1 |
英国（ソウル） | eu-west-2 |
フランクフルト（ドイツ） | eu-central-1 |
ムバイ（インド） | ap-outth-1 |
Tokyo, Japan | ap-northeast-1 |
香港 | ap-sautheast-1 |
オーストラリア、シドニー | ap-sautheast-2 |
これは、ストリーム プロセシング インスタンス自体にのみ適用されます。ストリーム プロセッサは引き続き、異なるクラウドプロバイダーでホストされているクラスター、または提供されている異なるリージョンでは、ストリーム プロセシング インスタンスと同じプロジェクト内にあるクラスターからの読み取りと書込みができます。
次のステップ
Atlas Stream Processing の主要概念の詳細については、以下をお読みください。