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

前提条件

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

• Stream Processing ワークスペース

• ストリーム プロセッサを作成および実行する atlasAdmin ロールを持つデータベースユーザー

• Atlas クラスター

Considerations

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

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

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

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

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

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

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

• 名前

• パイプライン
• デッド文字キュー

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

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

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

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

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

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

既存のストリーム プロセッサからサンプル結果の配列を mongosh を使用して STDOUT に返すには、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')
   }
}

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

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