Atlas Stream Processing を使い始める

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

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

1. Stream Processing ワークスペースに接続します。

   Stream Processing ワークスペースに関連付けられた接続文字列を使用して、mongoshを使用して接続します。

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

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

   3. ダイアログに表示される接続文字列をコピーします。形式は以下の通りで、 <atlas-stream-processing-url> は Stream Processing ワークスペースのURL、 <username> は atlasAdmin ロールを持つデータベースユーザーのユーザー名です。

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

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

      Enter キーを押して実行し、 Stream Processing ワークスペースに接続します。

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 はWindowsを使用してデータセットをバインドします。ストリームを連続する 10 秒間隔に分割する $tumblingWindow ステージの変数を定義します。

   つまり、例、$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（ストリーム プロセシング） を参照してください。

Stream Processing ワークスペースのストリーム プロセッサのリストで、ストリーム プロセッサの Start アイコンをクリックします。

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

sp.solarDemo.start()

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

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

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

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

   3. サイドバーで、 Database見出しの下のClustersをクリックします。

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

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

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

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

   3. サイドバーで、 Database見出しの下のData Explorerをクリックします。

      Data Explorerが表示されます。

   注意

   また、Clusters ページに go し、Shortcuts 見出しの下の 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
}

Stream Processing ワークスペースのストリーム プロセッサのリストで、ストリーム プロセッサの Delete（）アイコンをクリックします。

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

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

sp.solarDemo.drop()

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

sp.listStreamProcessors()