ストリームプロセッサの開発と管理

An Atlas Stream Processing ストリームプロセッサは、一意の名前を持つストリーム集計パイプラインのロジックをストリーミングデータに適用します。Atlas Stream Processing は、各ストリームプロセッサの定義を永続ストレージに保存し、再利用できるようにします。指定したストリームプロセッサは、その定義が保存されている Stream Processing ワークスペースでのみ使用できます。

Atlas Stream Processing がリソース使用量に応じてプロセッサの階層を自動的に調整するには、垂直オートスケーリングを有効にします。

インターフェース

前提条件

ストリームプロセッサを作成および管理するには、次のものが必要です。

Stream Processing ワークスペース
ストリームプロセッサを作成および実行するatlasAdmin ロールを持つデータベースユーザー
Atlas クラスター

Considerations

多くのストリームプロセッサコマンドでは、メソッド呼び出しで関連するストリームプロセッサの名前を指定する必要があります。次のセクションで説明される構文では、厳密に英数字の名前を使用することを前提としています。ストリームプロセッサの名前にハイフン（-）や完全停止（.）など、英数字以外の文字が含まれている場合は、名前を角括弧（[]）とdouble引用符（""）で囲む必要があります。メソッド呼び出し（sp.["special-name-stream"].stats() と同様）。

ストリームプロセッサを対話的に作成する

ストリームプロセッサを作成する

ストリームプロセッサを起動する

注意

Atlas Stream Processing は、45 日以上 stopped 状態にあるストリームプロセッサの内部状態を破棄します。そのようなプロセッサを開始すると、そのプロセッサは動作し、最初の実行と同一の統計をレポートします。

ストリームプロセッサの停止

注意

ストリームプロセッサを変更するには

既存のストリームプロセッサの次の要素を変更できます。

ストリームプロセッサを変更するには、次の手順に従います。

デフォルトでは、変更されたプロセッサは最後のチェックポイントから復元されます。あるいは、resumeFromCheckpoint=false を設定することもできます。この場合、プロセッサは要約統計のみを保持します。ウィンドウが開いているプロセッサーを変更した場合、ウィンドウは更新されたパイプラインで全て再計算されます。

注意

Operator（is、contains などのマッチャー式が含まれる）を使用して、ストリームプロセッサステートが失敗した場合のアラートを構成したストリームプロセッサの名前を変更すると、Atlas は、マッチャー式がその新しい名前と一致しない場合、名前が変更されたストリームプロセッサに対してアラートをトリガーしません。名前が変更されたストリームプロセッサを監視するには、アラートを再構成してください。

制限

デフォルト設定 resumeFromCheckpoint=true が有効になっている場合、以下の制限が適用されます。

$source ステージは変更できません。
ウィンドウの間隔は変更できません。
ウィンドウを削除することはできません。
ウィンドウを使用してパイプラインを変更できるのは、そのウィンドウの内部パイプラインに $group または $sort ステージのいずれかが含まれている場合のみです。
既存のウィンドウタイプを変更することはできません。たとえば、$tumblingWindow から $hoppingWindow への変更、またはその逆の変更はできません。
ウィンドウを持つプロセッサは、ウィンドウの再計算の結果として、一部のデータを再処理する可能性があります。
変更操作後は、演算子ごとの統計は保持されません。

1 つのストリームプロセッサのフェイルオーバーを開始

ストリームプロセッサの削除

使用可能なストリームプロセッサを一覧表示する

ワークスペースのデフォルトを一覧表示する

ストリームプロセッサからのサンプル

既存のストリームプロセッサからサンプル結果の配列をを使用してSTDOUT mongoshに返すには、メソッドを使用します。例、次のコマンドは、sp.processor.sample() という名前のストリームproc01 プロセッサからサンプリングしています。

sp.proc01.sample()

このコマンドは、 CTRL-Cを使用してキャンセルするまで、または返されたサンプルのサイズが累積40 MB になるまで継続的に実行されます。ストリームプロセッサは、サンプル内の無効なドキュメントを次の形式の_dlqMessageドキュメントで報告します。

 {
   _dlqMessage: {
     errInfo: {
    reason: "<reasonForError>"
     },
     doc: {
    _id: ObjectId('<group-id>'),
    ...
     },
     processorName: '<procName>',
     workspaceName: '<workspaceName>',
     dlqTime: ISODate('2024-09-19T20:04:34.263+00:00')
   }
}

これらのメッセージを使用して、デッドレターキューコレクションを定義せずにデータのクリーンアップの問題を診断できます。

ストリームプロセッサの統計情報を表示する

注意

Atlas Stream Processing がストリームプロセッサについてレポートするメトリクスの説明については、「ストリームプロセッサメトリクス」を参照してください。

戻る

VPC接続の管理

集計演算子

ストリーム プロセッサの開発と管理

前提条件

Considerations

ストリーム プロセッサを対話的に作成する

ストリーム プロセッサを作成する

ストリーム プロセッサを起動する

ストリーム プロセッサの停止