Atlas Stream Processing の概要

ソース

ストリーム プロセッサは、Atlas Stream Processing が接続されているストリーミングデータのソースからドキュメントを取り込むことで始まります。これらは、Apache Kafka などのサーバー システムや、Atlas の読み取り/書き込み操作によって生成されるようなデータベース変更ストリームでもあります。これらの入力は有効な json または ejson ドキュメントである必要があります。$source ステージがドキュメントを取り込むと、必要に応じてそのドキュメントにMongoDB集計を適用して変換できます。

Atlas Stream Processing は、ストリーミングソースからのデータを取り込むだけでなく、接続された Atlas クラスターのデータを結合するために、HTTPS requests と $lookup 操作のデータを使用してドキュメントを強化することもサポートされています。

パイプライン

ストリーム プロセッサは、取り込まれたデータを変換し、価値のあるインサイトを抽出するために、 MongoDBの標準的な 集計演算子とステージ に加えて、 集計パイプラインステージ と 集計演算子 を活用します。Atlas Stream Processing は、処理できないドキュメントを deadline 文字キュー に書き込むことができます。

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

Sinks

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