クイック スタート:Azure CLI を使用して Azure Stream Analytics ジョブを作成する

  • [アーティクル]

このクイックスタートでは、Azure CLI を使用して、温度が 27 度を超えるリアルタイム センサー メッセージをフィルター処理する、Stream Analytics ジョブを定義します。 Stream Analytics ジョブは、IoT Hub からデータを読み取り、データを変換し、出力データを BLOB ストレージ内のコンテナーに書き込みます。 このクイック スタートで使用される入力データは、Raspberry Pi オンライン シミュレーターによって生成されます。

開始する前に

Azure サブスクリプションをお持ちでない場合は、開始する前に Azure 無料アカウントを作成してください。

前提条件

  • Azure Cloud Shell で Bash 環境を使用します。 詳細については、「Azure Cloud Shell の Bash のクイックスタート」を参照してください。

  • CLI リファレンス コマンドをローカルで実行する場合、Azure CLI をインストールします。 Windows または macOS で実行している場合は、Docker コンテナーで Azure CLI を実行することを検討してください。 詳細については、「Docker コンテナーで Azure CLI を実行する方法」を参照してください。

    • ローカル インストールを使用する場合は、az login コマンドを使用して Azure CLI にサインインします。 認証プロセスを完了するには、ターミナルに表示される手順に従います。 その他のサインイン オプションについては、Azure CLI でのサインインに関するページを参照してください。

    • 初回使用時にインストールを求められたら、Azure CLI 拡張機能をインストールします。 拡張機能の詳細については、Azure CLI で拡張機能を使用する方法に関するページを参照してください。

    • az version を実行し、インストールされているバージョンおよび依存ライブラリを検索します。 最新バージョンにアップグレードするには、az upgrade を実行します。

  • リソース グループを作成します。 Azure リソースはすべてリソース グループにデプロイする必要があります。 リソース グループを使用すると、関連する Azure リソースを整理して管理できます。

    このクイックスタートでは、次の az group create コマンドを使用して、streamanalyticsrg という名前のリソース グループを eastus の場所に作成します。

    az group create --name streamanalyticsrg --location eastus

入力データを準備する

Stream Analytics ジョブを定義する前に、ジョブの入力に使用されるデータを準備します。 次の Azure CLI コマンドは、ジョブに必要な入力データを準備します。

  1. az iot hub create コマンドを使用して、IoT ハブを作成します。 この例では、MyASAIoTHub という名前の IoT ハブを作成します。 IoT Hub 名はグローバルに一意である必要があるため、既に使用されている場合は名前を変更する必要があります。 サブスクリプションで Free レベルを使用できる場合は、SKU を F1 に設定して Free レベルを使用します。 そうでない場合は、次に低いレベルを選択します。

    iotHubName=MyASAIoTHub
az iot hub create --name $iotHubName --resource-group streamanalyticsrg --sku S1

    IoT ハブが作成されたら、az iot hub connection-string show コマンドを使用して IoT Hub の接続文字列を取得します。 接続文字列全体をコピーして保存します。 IoT Hub を入力として Stream Analytics ジョブに追加するときに使用します。

    az iot hub connection-string show --hub-name $iotHubName

  2. az iothub device-identity create コマンドを使用して、デバイスを IoT ハブに追加します。 この例では、MyASAIoTDevice という名前のデバイスを作成します。

    az iot hub device-identity create --hub-name $iotHubName --device-id "MyASAIoTDevice"

  3. az iot hub device-identity connection-string show コマンドを使用してデバイスの接続文字列を取得します。 Raspberry Pi シミュレーターを作成するときに備えて、接続文字列全体をコピーして保存します。

    az iot hub device-identity connection-string show --hub-name $iotHubName --device-id "MyASAIoTDevice" --output table

    出力例:

    HostName=MyASAIoTHub.azure-devices.net;DeviceId=MyASAIoTDevice;SharedAccessKey=a2mnUsg52+NIgYudxYYUNXI67r0JmNubmfVafojG8=

BLOB ストレージ アカウントの作成

次の Azure CLI コマンドでは、ジョブ出力に使用される BLOB ストレージ アカウントを作成します。

  1. az storage account create コマンドで汎用のストレージ アカウントを作成します。 汎用のストレージ アカウントは、BLOB、ファイル、テーブル、およびキューという 4 つのサービスすべてに使用できます。

    storageAccountName="asatutorialstorage$RANDOM"
az storage account create \
    --name $storageAccountName \
    --resource-group streamanalyticsrg \
    --location eastus \
    --sku Standard_ZRS \
    --encryption-services blob

  2. az storage account keys list コマンドを実行して、ストレージ アカウントのキーを取得します。

    key=$(az storage account keys list -g streamanalyticsrg -n $storageAccountName --query "[0].value" -o tsv)
echo $key

    重要

    Azure ストレージ アカウントのアクセス キーをメモします。 このキーは、このクイックスタートの後の方で使用します。

  3. BLOB を格納する state という名前のコンテナーは、az storage container create コマンドで作成します。 ストレージ アカウント キーを使用して、コンテナーの作成操作を承認します。 Azure CLI を使用したデータ操作の承認について詳しくは、「Azure CLI を使用して BLOB またはキュー データへのアクセスを承認する」を参照してください。

    az storage container create \
    --account-name $storageAccountName \
    --name state \
    --account-key $key \
    --auth-mode key

Stream Analytics のジョブの作成

az stream-analytics job create コマンドを使用して、Stream Analytics ジョブを作成します。

az stream-analytics job create \
    --job-name "streamanalyticsjob" \
    --resource-group "streamanalyticsrg" \
    --location "eastus" \
    --output-error-policy "Drop" \
    --out-of-order-policy "Drop" \
    --order-max-delay 5 \
    --arrival-max-delay 16 \
    --data-locale "en-US"

ジョブへの入力を構成する

az stream-analytics input コマンドレットを使用して、ジョブへの入力を追加します。 このコマンドレットは、パラメーターとして、ジョブ名、ジョブ入力名、リソース グループ名、および入力プロパティを JSON 形式で受け取ります。 この例では、入力として IoT Hub を作成します。

重要

  • IOT HUB ACCESS KEY を、保存した IOT Hub 接続文字列の Shared Access Key の値に置き換えます。 たとえば、IOT Hub 接続文字列が HostName=MyASAIoTHub.azure-devices.net;SharedAccessKeyName=iothubowner;SharedAccessKey=xxxxxxxxxxxxxx= の場合、Shared Access Key の値は xxxxxxxxxxxxxx= です。 値を置き換える際に、" (二重引用符) の \ (エスケープ) 文字を削除しないように注意してください。
  • MyASAIoTHub 以外の名前を使用した場合は、次のコマンドで iotHubNamespace の値を更新してください。 echo $iotHubName を実行して、IoT Hub の名前を確認します。
az stream-analytics input create \
    --properties "{\"type\":\"Stream\",\"datasource\":{\"type\":\"Microsoft.Devices/IotHubs\",\"properties\":{\"consumerGroupName\":\"\$Default\",\"endpoint\":\"messages/events\",\"iotHubNamespace\":\"MyASAIoTHub\",\"sharedAccessPolicyKey\":\"IOT HUB ACCESS KEY\",\"sharedAccessPolicyName\":\"iothubowner\"}},\"serialization\":{\"type\":\"Json\",\"encoding\":\"UTF8\"}}" \
    --input-name "asaiotinput" \
    --job-name "streamanalyticsjob" \
    --resource-group "streamanalyticsrg"

ジョブへの出力を構成する

az stream-analytics output create コマンドレットを使用して、ジョブへの出力を追加します。 このコマンドレットは、ジョブ名、ジョブ出力名、リソース グループ名、JSON 形式のデータ ソース、シリアル化の種類をパラメーターとして受け取ります。

重要

STORAGEACCOUNTNAME> は Azure Storage アカウントの名前に、STORAGEACCESSKEY> はストレージ アカウントのアクセス キーで置き換えます。 これらの値をメモしていない場合は、コマンド echo $storageAccountName および echo $key を実行して値を取得します。 値を置き換える際に、" (二重引用符) の \ (エスケープ) 文字を削除しないように注意してください。

az stream-analytics output create \
    --job-name streamanalyticsjob \
    --datasource "{\"type\":\"Microsoft.Storage/Blob\",\"properties\":{\"container\":\"state\",\"dateFormat\":\"yyyy/MM/dd\",\"pathPattern\":\"{date}/{time}\",\"storageAccounts\":[{\"accountKey\":\"STORAGEACCESSKEY\",\"accountName\":\"STORAGEACCOUNTNAME\"}],\"timeFormat\":\"HH\"}}" \
    --serialization "{\"type\":\"Json\",\"properties\":{\"format\":\"Array\",\"encoding\":\"UTF8\"}}" \
    --output-name asabloboutput \
    --resource-group streamanalyticsrg

変換クエリを定義する

az stream-analytics transformation create コマンドレットを使用して、ジョブの変換を追加します。

az stream-analytics transformation create \
    --resource-group streamanalyticsrg \
    --job-name streamanalyticsjob \
    --name Transformation \
    --streaming-units "6" \
    --saql "SELECT * INTO asabloboutput FROM asaiotinput WHERE Temperature > 27"

IoT シミュレーターを実行する

  1. Raspberry Pi Azure IoT オンライン シミュレーターを開きます。

  2. 15 行目のプレースホルダーを、クイックスタートの最初に保存した Azure IoT Hub デバイス接続文字列 (IoT Hub 接続文字列ではない) 全体に置き換えます。

  3. [実行] を選択します。 お客様の IoT ハブに送信されているセンサー データとメッセージが出力に表示されます。

    Raspberry Pi Azure IoT Online Simulator

Stream Analytics ジョブを開始して出力をチェックする

az stream-analytics job start コマンドレットを使用して、ジョブを開始します。 このコマンドレットは、パラメーターとして、ジョブ名、リソース グループ名、出力開始モード、および開始時刻を受け取ります。 OutputStartMode には、JobStartTime、CustomTime、または LastOutputEventTime の値が使用できます。

次のコマンドレットを実行すると、ジョブが開始されたときに出力として True が返されます。

az stream-analytics job start \
    --resource-group streamanalyticsrg \
    --name streamanalyticsjob \
    --output-start-mode JobStartTime

数分待ってから、state BLOB コンテナーに出力ファイルが作成されていることを確認します。

Screenshot showing the output file in the State blob container.

ファイルをダウンロードして開き、次のようないくつかのエントリを表示します。

{
    "messageId": 229,
    "deviceId": "Raspberry Pi Web Client",
    "temperature": 31.85214010589595,
    "humidity": 60.278830289656284,
    "EventProcessedUtcTime": "2023-02-28T22:06:33.5567789Z",
    "PartitionId": 3,
    "EventEnqueuedUtcTime": "2023-02-28T22:05:49.6520000Z",
    "IoTHub": {
        "MessageId": null,
        "CorrelationId": null,
        "ConnectionDeviceId": "MyASAIoTDevice",
        "ConnectionDeviceGenerationId": "638132150746523845",
        "EnqueuedTime": "2023-02-28T22:05:49.6520000Z",
        "StreamId": null
    }
}

リソースをクリーンアップする

リソース グループを削除します。これにより、Stream Analytics ジョブ、IoT Hub、Azure Storage アカウントを含むリソース グループ内のすべてのリソースが削除されます。

az group delete \
    --name streamanalyticsrg \
    --no-wait

次のステップ

このクイックスタートでは、Azure CLI を使用して、簡単な Stream Analytics ジョブをデプロイしました。 Azure portal と Visual Studio を使用して Stream Analytics ジョブをデプロイすることもできます。

その他の入力ソースの構成やリアルタイム検出の実行について学習するには、次の記事に進んでください。

Azure Stream Analytics を使用したリアルタイムの不正行為の検出