Azure Stream Analytics を使用したリアルタイム IoT データストリームの処理

[アーティクル]
08/21/2023

この記事では、モノのインターネット (IoT) デバイスからデータを収集するストリーム処理ロジックの作成方法について学習します。実際のモノのインターネット (IoT) ユースケースを使用して、迅速で経済的なソリューションを構築する方法を紹介します。

前提条件

無料の Azure サブスクリプションを作成する。
GitHub からサンプルクエリとデータファイルをダウンロードする。

シナリオ

Contoso は工業オートメーションの領域で活動する会社で、自社の製造工程を自動化しています。この工場の機械には、リアルタイムでデータのストリームを生成することのできるセンサーがあります。このシナリオにおいて、生産現場マネージャーは、センサーデータからリアルタイムの詳細情報を取得し、パターンを見つけて、それらに対処したいと考えています。センサーデータに対して Stream Analytics クエリ言語 (SAQL) を使用して、データの受信ストリームから興味深いパターンを検出できます。

この例では、データは Texas Instruments 社のセンサータグデバイスから生成されます。データのペイロードは、次のサンプルスニペットに示すように JSON 形式です。

{
    "time": "2016-01-26T20:47:53.0000000",  
    "dspl": "sensorE",  
    "temp": 123,  
    "hmdt": 34  
}

実際のシナリオでは、何百ものこのようなセンサーがストリームとしてイベントを生成することになります。ゲートウェイデバイスがコードを実行し、これらのイベントを Azure Event Hubs または Azure IoT Hub にプッシュできれば理想的です。 Stream Analytics ジョブでそれらのイベントを Event Hubs または IoT Hubs から取り込み、そのストリームに対してリアルタイム分析クエリを実行します。結果はその後、いずれかのサポートされている出力に送信することになります。

このガイドでは使いやすさを考えて、現実のセンサータグデバイスからキャプチャしたサンプルデータファイルを用意しています。このサンプルデータに対してクエリを実行し、結果を確認できます。以降のチュートリアルでは、ジョブを入力と出力に関連付け、それらを Azure サービスにデプロイする方法を学習します。

Stream Analytics のジョブの作成

Azure Portal に移動します。
左側のナビゲーションメニューで、[すべてのサービス] を選択し、[分析] を選択し、[Stream Analytics jobs] (Stream Analytics ジョブ) の上にマウスポインターを置き、[作成] を選択します。
[新しい Stream Analytics ジョブ] ページで、次の手順のようにします。
1. [サブスクリプション] で、自分の Azure サブスクリプションを選択します。
2. [リソースグループ] で、既存のリソースグループを選ぶか、リソースグループを作成します。
3. [名前] に、Stream Analytics ジョブの一意の名前を入力します。
4. Stream Analytics ジョブを作成するリージョンを選びます。処理速度を向上させ、コストを削減するために、リソースグループとすべてのリソースに同じ場所を使用してください。
5. [Review + create](レビュー + 作成) を選択します。
[確認および作成] ページで、設定を確認し、 [作成] を選択します。
デプロイが成功したら、[リソースに移動] を選んで Stream Analytics ジョブの [Stream Analytics ジョブ] ページに移動します。

Azure Stream Analytics クエリの作成

ジョブが作成されたら、クエリを記述します。入力または出力をジョブに接続しなくても、サンプルデータに対してクエリをテストできます。

GitHub から HelloWorldASA-InputStream.json をダウンロードします。
Azure portal の [Azure Stream Analytics ジョブ] ページで、左側のメニューから [ジョブトポロジ] の下の [クエリ] を選択します。
[Upload sample input] (サンプル入力のアップロード) を選択し、ダウンロードした HelloWorldASA-InputStream.json ファイルを選択して、[OK] を選択します。
データのプレビューが 入力プレビュー テーブルに自動的に設定されることに注意してください。

クエリ: 生データのアーカイブ

クエリの最も単純な形式は、すべての入力データを指定された出力にアーカイブするパススルークエリです。このクエリは、新しい Azure Stream Analytics ジョブで設定された既定のクエリです。

[クエリ] ウィンドウで、次のクエリを入力し、ツールバーの [Test query] (クエリのテスト) を選択します。
```
SELECT
    *
INTO
    youroutputalias
FROM
    yourinputalias
```
下部ウィンドウの [Test results] (テスト結果) タブで結果を確認します。

クエリ: 条件に基づいたデータのフィルター処理

条件に基づいて結果をフィルター処理するようにクエリを更新してみましょう。たとえば、次のクエリは、sensorA からのイベントを表示します。

次のサンプルを使用してこのクエリを更新します。

SELECT 
    time,
    dspl AS SensorName,
    temp AS Temperature,
    hmdt AS Humidity
INTO
   youroutputalias
FROM
    yourinputalias
WHERE dspl='sensorA'

[Test query] (クエリのテスト) を選択してクエリの結果を確認します。

クエリ: ビジネスワークフローをトリガーするアラート

クエリについて、もう少し詳しく説明します。あらゆる種類のセンサーを対象に、30 秒間隔で平均温度を監視し、平均温度が 100 度を超える場合にのみ結果を表示するのであれば、

クエリを次のように更新します。

SELECT 
    System.Timestamp AS OutputTime,
    dspl AS SensorName,
    Avg(temp) AS AvgTemperature
INTO
   youroutputalias
FROM
    yourinputalias TIMESTAMP BY time
GROUP BY TumblingWindow(second,30),dspl
HAVING Avg(temp)>100

[Test query] (クエリのテスト) を選択してクエリの結果を確認します。

ご覧のように、結果に含まれるのは 245 行のみで、平均温度が 100 度を超えるセンサーの名前が一覧表示されます。このクエリでは、センサー名である dspl 別に、30 秒のタンブリングウィンドウでイベントのストリームをグループ化しています。一時的なクエリでは、時間の進み方を指定する必要があります。ここでは、一時的な計算すべてに時間を関連付けるため、TIMESTAMP BY 句を使用して OUTPUTTIME 列を指定しました。詳細については、時間管理とウィンドウ関数に関するページを参照してください。

クエリ: イベントがないことを検出する

入力イベントがないことを検出するためのクエリは、どのように記述すればよいのでしょうか。センサーが最後にデータを送信してから 5 秒間イベントを送信しなかったタイミングを見つけましょう。