Atlas Stream Processing を使い始める

Atlas UIでストリーム プロセッサを作成するには、Atlasプロジェクトの Stream Processing ページに移動し、ストリーム処理インスタンスの ペインで [Configure] をクリックします。次に、ビジュアル ビルダまたはJSONエディターのどちらかを使用して、solarDemo という名前のストリーム プロセッサを構成します。

mongoshで次のコマンドを実行して、 という名前の永続ストリームsolarDemo プロセッサを作成します。

1. Atlas Stream Processing インスタンスに接続します。

   Atlas Stream Processing インスタンスに関連付けられた接続文字列を使用して接続し、 mongosh を使用して接続します。

   1. Atlas Stream Processing インスタンスの ペインで、 Connectをクリックします。

   2. Connect to your instance ダイアログで、Shellタブを選択します。

   3. ダイアログに表示される接続文字列をコピーします。形式は次の<atlas-stream-processing-url> で、<username> はストリーム処理インスタンスのURL 、atlasAdmin は ロールを持つデータベースユーザーのユーザー名です。

      mongosh "mongodb://<atlas-stream-processing-url>/"
      --tls --authenticationDatabase admin --username <username>
      --password <password>

   4. 接続文字列をターミナルに貼り付け、<password> プレースホルダーをユーザーの認証情報に置き換えます。

      Enter キーを押して実行し、ストリーム処理インスタンスに接続します。

2. $sourceステージを構成します。

   sample_stream_solar ソースからデータを取り込む $source ステージの変数を定義します。

   let s = {
     $source: {
       connectionName: "sample_stream_solar"
     }
   }

3. $groupステージを構成します。

   $group ステージの変数を定義し、group_id に従って最大温度と各ソート デバイスの平均、最大、最小出力を出力します。

   let g = {
     $group: {
       _id: "$group_id",
       max_temp: {
         $max: "$obs.temp"
       },
       avg_watts: {
          $avg: "$obs.watts"
       },
       max_watts: {
         $max: "$obs.watts"
       },
       min_watts: {
         $min: "$obs.watts"
       }
     }
   }

4. $tumblingWindowステージを構成します。

   $groupストリーミングデータで などのアキュムレーションを実行するために、Atlas Stream Processing は ウィンドウ$tumblingWindow 10を使用してデータセットをバインドします。ストリームを連続する 秒間隔に分割する ステージの変数を定義します。

   つまり、例、$group ステージが max_watts の 値を計算する場合、過去 10 秒に特定の group_id が取り込まれたすべてのドキュメントの obs.watts 値から最大値が抽出されます。

    let t = {
      $tumblingWindow: {
        interval: {
          size: NumberInt(10),
          unit: "second"
        },
        pipeline: [g]
     }
   }

5. $mergeステージを構成します。

   接続された Atlas クラスターの solarDbデータベース内の solarColl という名前のコレクションに処理されたストリーミングデータを書込む $merge ステージの変数を定義します。

   let m = {
     $merge: {
       into: {
         connectionName: "mongodb1",
         db: "solarDb",
         coll: "solarColl"
       }
     }
   }

6. ストリーム プロセッサを作成します。

   sp.createStreamProcessor()メソッドを使用して、新しいストリーム プロセッサに名前を割り当て、その集計パイプラインを宣言します。$group ステージは$tumblingWindow のネストされたパイプラインに属しており、プロセッサパイプライン定義にこれを含めることはできません。

   sp.createStreamProcessor("solarDemo", [s, t, m])

   これにより、以前に定義されたクエリを適用し、接続したクラスター上のsolarDbデータベースのsolarCollコレクションに処理されたデータを書込むsolarDemoという名前のストリーム プロセッサが作成されます。 ソート デバイスからの10秒間隔の観察から派生したさまざまな測定値を返します。

   Atlas Stream Processing が保管中のデータベースに書き込む方法の詳細については、$merge（ストリーム プロセシング） を参照してください。

ストリーム処理インスタンスのストリーム プロセッサのリストで、ストリーム プロセッサの Start アイコンをクリックします。

sp.processor.start()mongoshで メソッドを使用する:

sp.solarDemo.start()

ストリーム プロセッサが Atlas クラスターにデータを書き込んだことを確認するには、次の手順を実行します。

1. Atlas で、プロジェクトの [Clusters] ページに移動します。

   警告: ナビゲーションの改善が進行中

   現在、新しく改善されたナビゲーション エクスペリエンスを展開しています。次の手順が Atlas UIのビューと一致しない場合は、プレビュー ドキュメントを参照してください。

   1. まだ表示されていない場合は、希望するプロジェクトを含む組織を選択しますナビゲーション バーのOrganizationsメニュー

   2. まだ表示されていない場合は、ナビゲーション バーのProjectsメニューから目的のプロジェクトを選択します。

   3. まだ表示されていない場合は、サイドバーの [Clusters] をクリックします。

      [ Clusters （クラスター） ] ページが表示されます。

2. クラスターの [Browse Collections] ボタンをクリックします。

   Data Explorerが表示されます。

3. MySolarコレクションを表示します。

プロセッサがアクティブであることを確認するには、sp.processor.stats() mongoshの メソッドを使用します。

sp.solarDemo.stats()

このメソッドは、solarDemoストリーム プロセッサの運用統計を報告します。

また、sp.processor.sample() のmongosh メソッドを使用して、ターミナルに処理されたドキュメントのサンプルを返すこともできます。

sp.solarDemo.sample()

{
  _id: 10,
  max_temp: 16,
  avg_watts: 232,
  max_watts: 414,
  min_watts: 73
}

ストリーム処理インスタンスのストリーム プロセッサのリストで、ストリーム プロセッサのDelete （ ）アイコンをクリックします。

表示される確認ダイアログで、ストリーム プロセッサの名前（solarDemo）を入力して削除することを確認し、[Delete] をクリックします。

sp.processor.drop()mongoshsolarDemoを削除するには、 の メソッドを使用します。

sp.solarDemo.drop()

solarDemoを削除したことを確認するには、 メソッドを使用して使用可能なすべてのストリームsp.listStreamProcessors() プロセッサを一覧表示します。

sp.listStreamProcessors()