Atlas Stream Processing

ソース

Stream processors begin by ingesting documents from sources of streaming data to which Atlas Stream Processing is connected. These can be broker systems like Apache Kafka, or database change streams such as those generated by Atlas read/write operations. These inputs must be valid json or ejson documents. Once the $source stage ingests a document, you can apply MongoDB aggregation to that document to transform it as needed.

Atlas Stream Processing は、ストリーミングソースからのデータの取り込みに加えて、HTTPS リクエスト、接続された Atlas クラスターからのデータを結合する $lookup 操作、MongoDB Vector Search インデックスに対する $vectorSearch クエリのデータを使用してドキュメントを強化することもサポートしています。

パイプライン

ストリーム プロセッサは、集計パイプラインステージと集計演算子に加えて、標準のMongoDB スイートの集計演算子とステージを活用して、取り込んだデータを変換し、価値あるインサイトを抽出します。Atlas Stream Processing の集計パイプラインを定義する方法については、集計パイプラインのドキュメントをご覧ください。

Atlas Stream Processing は処理できないドキュメントをデッドレターキュー（DLQ）に書き込むことができます。

ドキュメントを再構築したり、フィールドを追加または削除したり、コレクションから情報を検索したりすることで、ドキュメントを増やします。Atlas Stream Processing では、Windowsを使用してイベントを収集し、任意の関数を実行することもできます。

Sinks

取り込まれたデータをプロセシングした後、ストリームプロセッサーはシンクに書き込むことで持続させます。Atlas Stream Processingは、異なるシンクタイプへの書き込みに$emitステージと$mergeステージを提供します。これらのステージは互いに排他的であり、各ストリームプロセッサーは1つのシンクステージしか持つことができません。パイプラインには、同じシンク接続内の異なるKafkaトピックまたはAtlasコレクションへのプロセシング済みドキュメントの書き込みのロジックを含めることができます。