IoT データを Azure Data Explorer にエクスポートする

[アーティクル]
03/05/2024

この記事では、Azure Data Explorer にデータを送信するようにデータエクスポートを構成する方法について説明します。

この機能を使用して、フィルター処理およびエンリッチ化された IoT データを IoT Central アプリケーションから連続エクスポートします。データエクスポートを使用すると、ウォームパスの分析情報、分析、およびストレージ用に、クラウドソリューションの他の部分にほぼリアルタイムで変更がプッシュされます。

たとえば、次のように操作できます。

テレメトリ、プロパティ変更、デバイスの接続性、デバイスのライフサイクル、デバイスのテンプレートのライフサイクル、および監査ログデータを、JSON 形式で、ほぼリアルタイムで継続的にエクスポートする
データストリームをフィルター処理して、カスタム条件に一致するデータをエクスポートする
デバイスからのカスタム値とプロパティ値を使用してデータストリームをエンリッチ化する
データストリームを変換して、形態とコンテンツを変更します。

ヒント

データエクスポートを有効にすると、その時点以降のデータのみが取得されます。より多くの履歴データを保持するには、データエクスポートを早い段階で有効にしてください。データエクスポートが無効になっていたときからのデータを手動でエクスポートするには、「IoT Central REST API を使ってデバイスのクエリを実行する方法」を参照してください。

Note

状況によっては、メッセージがエクスポートされるまでに最大 60 秒かかる場合があります。これは、IoT Central が基になる IoT ハブからメッセージを受信した時点から、メッセージが宛先エンドポイントに配信されるまでの時間です。

前提条件

データエクスポート機能を使用するには、データエクスポートのアクセス許可が必要です。

Azure Data Explorer のエクスポート先を設定する

Azure Data Explorer クラスターまたは Azure Synapse Data Explorer プールを使用できます。詳細については、Azure Synapse Data Explorer と Azure Data Explorer の違いに関するページを参照してください。

データは、IoT Central から Azure Data Explorer クラスターのデータベーステーブルにほぼリアルタイムでエクスポートされます。データはメッセージ本文に含まれ、UTF-8 としてエンコードされた JSON 形式です。 IoT Central に変換を追加して、テーブルスキーマに一致するデータをエクスポートすることができます。

Azure Data Explorer ポータルでエクスポートされたデータに対してクエリを実行するには、データベースに移動し、[クエリ] を選択します。

次のビデオでは、Azure Data Explorer にデータをエクスポートする方法について説明します。

接続オプション

Azure Data Explorer を使用すると、サービスプリンシパルまたはマネージド ID を使用して接続を構成できます。

セキュリティは、マネージド ID の方が優れています。その理由は次のとおりです。

IoT Central アプリケーションの接続文字列にリソースの資格情報が格納されません。
資格情報は、IoT Central アプリケーションの有効期間に自動的に関連付けられます。
マネージド ID では、セキュリティキーの定期的なローテーションが自動的に行われます。

現在、IoT Central では、システム割り当てマネージド ID が使用されます。

マネージド ID を構成する場合、構成には スコープ" と ロール が含まれます:

スコープでは、マネージド ID を使用できる場所を定義します。たとえば、スコープとして Azure リソースグループを使用できます。この場合、IoT Central アプリケーションとエクスポート先の両方が同じリソースグループに含まれている必要があります。
ロールでは、エクスポート先のサービスで IoT Central アプリケーションに付与されるアクセス許可を定義します。たとえば、IoT Central アプリケーションがイベントハブにデータを送信するには、マネージド ID に Azure Event Hubs のデータ送信者ロールの割り当てが必要です。

次のビデオでは、システム割り当てマネージド ID の詳細について説明します。

注意

BLOB ストレージにエクスポートするには、ビデオに示すように、ストレージアカウント共同作成者を使用しないでください。代わりにストレージ BLOB データ共同作成者ロールを使用してください。

この記事では、Azure CLI でマネージド ID を作成する方法について説明します。 Azure portal を使ってマネージド ID を作成することもできます。

エクスポート先となる既存の Azure Data Explorer データベースがない場合は、次の手順に従います。 Azure Data Explorer データベースを作成します。次の 2 つの選択肢があります。

新しい Azure Data Explorer クラスターとデータベースを作成します。詳細については、Azure Data Explorer のクイックスタートに関するページを参照してください。クラスターの URI と作成するデータベースの名前をメモしておきます。これらの値は次の手順で必要になります。
新しい Azure Synapse Data Explorer プールとデータベースを作成します。詳細については、Azure Data Explorer のクイックスタートに関するページを参照してください。プールの URI と作成するデータベースの名前をメモしておきます。これらの値は次の手順で必要になります。

IoT Central アプリケーションが Azure リソースにデータを安全にエクスポートできるようにするマネージド ID を構成するには、次の手順を実行します。

データベースへの接続に使用する IoT Central アプリケーションのマネージド ID を作成します。 Azure Cloud Shell を使用して次のコマンドを実行します。
```
az iot central app identity assign --name {your IoT Central app name} \
    --resource-group {resource group name} \
    --system-assigned
```
コマンドによる principalId と tenantId 出力をメモします。これらの値は、次の手順で使用します。

IoT Central アプリケーションからの接続を許可するデータベース権限を構成します。 Azure Cloud Shell を使用して次のコマンドを実行します。

az kusto database-principal-assignment create --cluster-name {name of your cluster} \
    --database-name {name of your database}    \
    --resource-group {resource group name} \
    --principal-assignment-name {name of your IoT Central application} \
    --principal-id {principal id from the previous step} \
    --principal-type App --role Admin \
    --tenant-id {tenant id from the previous step}

ヒント

Azure Synapse を使用している場合は、az synapse kusto database-principal-assignment を参照してください。

エクスポートするデータに適したスキーマを使用して、データベースにテーブルを作成します。次のクエリ例では、smartvitalspatch という名前のテーブルを作成します。詳しくは、「IoT Central アプリケーション内のデータをエクスポート用に変換する」を参照してください。
```
.create table smartvitalspatch (
  EnqueuedTime:datetime,
  Message:string,
  Application:string,
  Device:string,
  Simulated:boolean,
  Template:string,
  Module:string,
  Component:string,
  Capability:string,
  Value:dynamic
)
```
(オプション) Azure Data Explorer データベースへのデータの取り込みを高速化するには、次のようにします。
1. Azure Data Explorer クラスターの [構成] ページに移動します。次に、[ストリーミングインジェスト] オプションを有効にします。
2. 次のクエリを実行して、テーブルポリシーを変更してストリーミングインジェストを有効にします。
```
.alter table smartvitalspatch policy streamingingestion enable
```

IoT Central の [データエクスポート] ページで Azure Data Explorer をエクスポート先として作成するには:

[+ New destination](+ 新しい宛先) を選択します。
エクスポート先の種類として Azure Data Explorer を選択します。
Azure Data Explorer クラスターまたはプールの URL、データベース名、テーブル名を入力します。認可の種類として [システム割り当てマネージド ID] を選択します。

ヒント

スタンドアロンの Azure Data Explorer に使用されるクラスター URL の形式は https://<ClusterName>.<AzureRegion>.kusto.windows.net です。 Azure Synapse Data Explorer プールに使用されるクラスター URL の形式は https://<DataExplorerPoolName>.<SynapseWorkspaceName>.kusto.azuresynapse.net です。

宛先サービスにデータが到着しない場合は、Azure IoT Central アプリケーションからのデータエクスポートに関する問題のトラブルシューティングに関するページを参照してください。

エクスポート先となる既存の Azure Data Explorer データベースがない場合は、次の手順に従います。

Azure Data Explorer データベースを作成します。次の 2 つの選択肢があります。
- 新しい Azure Data Explorer クラスターとデータベースを作成します。詳細については、Azure Data Explorer のクイックスタートに関するページを参照してください。クラスターの URI と作成するデータベースの名前をメモしておきます。これらの値は次の手順で必要になります。
- 新しい Azure Synapse Data Explorer プールとデータベースを作成します。詳細については、Azure Data Explorer のクイックスタートに関するページを参照してください。プールの URI と作成するデータベースの名前をメモしておきます。これらの値は次の手順で必要になります。
IoT Central アプリケーションを Azure Data Explorer に接続するために使用できるサービスプリンシパルを作成します。 Azure Cloud Shell を使用して次のコマンドを実行します。
```
az ad sp create-for-rbac --skip-assignment --name "My SP for IoT Central" --scopes /subscriptions/<SubscriptionId>
```
コマンドの出力で、appId、password、tenant の値をメモしておきます。これらの値は次の手順で必要になります。
サービスプリンシパルをデータベースに追加するには、Azure Data Explorer ポータルに移動し、データベースに対して次のクエリを実行します。プレースホルダーを、前に書き留めた値に置き換えます。
```
.add database ['<YourDatabaseName>'] admins ('aadapp=<YourAppId>;<YourTenant>');
```
エクスポートするデータに適したスキーマを使用して、データベースにテーブルを作成します。次のクエリ例では、smartvitalspatch という名前のテーブルを作成します。詳しくは、「IoT Central アプリケーション内のデータをエクスポート用に変換する」を参照してください。
```
.create table smartvitalspatch (
  EnqueuedTime:datetime,
  Message:string,
  Application:string,
  Device:string,
  Simulated:boolean,
  Template:string,
  Module:string,
  Component:string,
  Capability:string,
  Value:dynamic
)
```
(オプション) Azure Data Explorer データベースへのデータの取り込みを高速化するには、次のようにします。
1. Azure Data Explorer クラスターの [構成] ページに移動します。次に、[ストリーミングインジェスト] オプションを有効にします。
2. 次のクエリを実行して、テーブルポリシーを変更してストリーミングインジェストを有効にします。
```
.alter table smartvitalspatch policy streamingingestion enable
```

IoT Central の [データエクスポート] ページで Azure Data Explorer をエクスポート先として作成するには:

[+ New destination](+ 新しい宛先) を選択します。
エクスポート先の種類として Azure Data Explorer を選択します。
Azure Data Explorer クラスターまたはプールの URL、データベース名、テーブル名を入力します。次の表は、承認に使用するサービスプリンシパルの値を示しています。

サービスプリンシパルの値宛先の構成

appId ClientID

tenant テナント ID

password クライアントシークレット

ヒント

スタンドアロンの Azure Data Explorer に使用されるクラスター URL の形式は https://<ClusterName>.<AzureRegion>.kusto.windows.net です。 Azure Synapse Data Explorer プールに使用されるクラスター URL の形式は https://<DataExplorerPoolName>.<SynapseWorkspaceName>.kusto.azuresynapse.net です。

サービスプリンシパルの値	宛先の構成
appId	ClientID
tenant	テナント ID
password	クライアントシークレット

データエクスポートの設定

データのエクスポート先を準備したので、IoT Central アプリケーションでデータのエクスポートを設定します。

ご使用の IoT Central アプリケーションにサインインします。
左側のペインで、[データのエクスポート] を選択します。

ヒント

左側のペインに [データのエクスポート] が表示されない場合は、アプリでデータエクスポートを構成するアクセス許可がありません。データエクスポートの設定について、管理者に問い合わせてください。
[+ 新しいエクスポート] を選択します。
新しいエクスポートの表示名を入力し、データエクスポートが [有効] になっていることを確認します。

エクスポートするデータの種類を選択します。次の表は、サポートされるデータエクスポートの型の一覧を示します。

データの種類	説明	データ形式
テレメトリ	デバイスからのテレメトリメッセージがほぼリアルタイムでエクスポートされます。エクスポートされた各メッセージには、元のデバイスメッセージの完全な内容が正規化されて含まれます。	テレメトリメッセージの形式
プロパティ変更	デバイスとクラウドのプロパティに対する変更がほぼリアルタイムでエクスポートされます。読み取り専用のデバイスプロパティでは、報告された値に対する変更がエクスポートされます。読み取り/書き込みプロパティの場合、報告された値と必要な値の両方がエクスポートされます。	プロパティ変更メッセージの形式
デバイスの接続性	デバイスの接続イベントと非接続イベントをエクスポートします。	デバイス接続メッセージの形式
デバイスのライフサイクル	デバイスの登録、削除、プロビジョニング、有効化、無効化、displayNameChanged、および deviceTemplateChanged のイベントをエクスポートします。	デバイスのライフサイクル変更メッセージの形式
デバイステンプレートのライフサイクル	created、updated、および deleted など、発行されたデバイステンプレートの変更をエクスポートします。	デバイステンプレートのライフサイクル変更メッセージの形式
監査ログ	アプリケーション内のエンティティに対してユーザーが開始した更新のログ。詳細については、「監査ログを使用して IoT Central アプリケーションのアクティビティを追跡する」を参照してください	監査ログメッセージの形式

必要に応じて、フィルターを追加して、エクスポートするデータの量を減らします。使用できるフィルターの種類は、データエクスポートの種類ごとに異なります。

データの種類	使用可能なフィルター
テレメトリ	デバイス名、デバイス ID、デバイステンプレート、デバイスがシミュレートされているかどうかでフィルター処理しますフィルター条件を満たすテレメトリのみが含まれるようにストリームをフィルター処理しますフィルター条件に一致するプロパティを持つデバイスのテレメトリのみが含まれるようにストリームをフィルター処理しますフィルター条件を満たし、"メッセージプロパティ" を含むテレメトリのみが含まれるようにストリームをフィルター処理します。 "メッセージプロパティ" ("アプリケーションプロパティ" とも呼ばれます) は、各テレメトリメッセージのキーと値のペアのバッグとして送信されます。メッセージプロパティフィルターを作成するには、検索するメッセージプロパティキーを入力し、条件を指定します。指定したフィルター条件に一致するプロパティを持つテレメトリメッセージのみがエクスポートされます。アプリケーションプロパティの詳細については、IoT Hub のドキュメントを参照してください
プロパティ変更	デバイス名、デバイス ID、デバイステンプレート、デバイスがシミュレートされているかどうかでフィルター処理しますフィルター条件を満たすプロパティ変更のみが含まれるようにストリームをフィルター処理します
デバイスの接続性	デバイス名、デバイス ID、デバイステンプレート、組織、デバイスがシミュレートされているかどうかでフィルター処理しますフィルター条件に一致するプロパティを持つデバイスの変更のみが含まれるようにストリームをフィルター処理します
デバイスのライフサイクル	デバイス名、デバイス ID、デバイステンプレート、デバイスがプロビジョニング、有効化、またはシミュレートされているかどうかでフィルター処理しますフィルター条件に一致するプロパティを持つデバイスの変更のみが含まれるようにストリームをフィルター処理します
デバイステンプレートのライフサイクル	デバイステンプレートでフィルター処理します
監査ログ	該当なし

必要に応じて、余分なキーと値のペアのメタデータを使用して、エクスポートされたメッセージを強化します。次のエンリッチメントは、テレメトリ、プロパティ変更、デバイスの接続性、デバイスのライフサイクルの各データエクスポートの種類で使用できます。
- カスタム文字列: 各メッセージにカスタムの静的文字列を追加します。任意のキーを入力し、任意の文字列値を入力します。
- プロパティ: 各メッセージに追加されます。
  - デバイス名、デバイステンプレート名、有効、組織、プロビジョニング済み、シミュレート済みなどのデバイスメタデータ。
  - 各メッセージに対してデバイスから報告された現在のプロパティまたはクラウドプロパティの値。エクスポートされたメッセージが、指定したプロパティを持たないデバイスからのものである場合、エクスポートされたメッセージにはそのエンリッチメントがありません。

エクスポート先を構成します。

[+ Destination]\(+ エクスポート先\) を選択し、既に作成してあるエクスポート先を追加するか [新規作成する] を選択します。
エクスポート前にデータを変換するには、[+ 変換] を選択します。詳しくは、「IoT Central アプリケーション内のデータをエクスポート用に変換する」を参照してください。
[+ Destination]\(+ エクスポート先\) を選択して、1 回のエクスポートに最大 5 つのエクスポート先を追加します。
エクスポートの設定が完了したら、[保存] を選択します。数分後に、エクスポート先にデータが表示されます。

エクスポートの監視

エクスポートの状態は、IoT Central の [Data export]\(データのエクスポート\) ページで確認できます。 Azure Monitor を使用して、エクスポートするデータの量とエクスポートエラーを確認することもできます。エクスポートとデバイスの正常性のメトリックには、Azure portal 内のグラフ、REST API、または PowerShell や Azure CLI のクエリを使ってアクセスできます。現時点では、Azure Monitor で次のデータエクスポートメトリックを監視できます。

フィルター適用前のエクスポート対象受信メッセージの数。
フィルターを通過したメッセージの数。
宛先に正常にエクスポートされたメッセージの数。
見つかったエラーの数。

詳細については、「アプリケーションの正常性を監視する」を参照してください。

データ形式

以降のセクションでは、エクスポートデータの形式について説明します。

テレメトリ形式

エクスポートされた各メッセージには、メッセージ本文でデバイスから送信された、完全なメッセージが正規化された形式で含まれます。メッセージは JSON 形式で、UTF-8 としてエンコードされます。各メッセージに含まれる情報は次のとおりです。

applicationId: IoT Central アプリケーションの ID。
messageSource: メッセージのソース - telemetry。
deviceId: テレメトリメッセージを送信したデバイスの ID。
schema: ペイロードスキーマの名前とバージョン。
templateId: デバイスに割り当てられているデバイステンプレートの ID。
enqueuedTime: IoT Central がこのメッセージを受信した時刻。
enrichments: エクスポートに設定されたエンリッチメント。
module: このメッセージを送信した IoT Edge モジュール。このフィールドは、メッセージが IoT Edge モジュールから送信された場合にのみ表示されます。
component: このメッセージを送信したコンポーネント。このフィールドは、メッセージで送信された機能がデバイステンプレートでコンポーネントとしてモデル化されている場合にのみ表示されます。
messageProperties: デバイスによってメッセージと共に送信されるその他のプロパティ。これらのプロパティはアプリケーションプロパティと呼ばれることもあります。詳細については IoT Hub のドキュメントを参照してください。

メッセージプロパティ

テレメトリメッセージには、テレメトリペイロードに加え、メタデータのプロパティが含まれています。前のスニペットは、deviceId や enqueuedTime など、システムメッセージの例を示しています。システムメッセージプロパティの詳細については、「D2C IoT Hub メッセージのシステムプロパティ」を参照してください。

テレメトリメッセージにカスタムメタデータを追加する必要がある場合、テレメトリメッセージにプロパティを追加できます。たとえば、デバイスでメッセージが作成されるとき、タイムスタンプを追加する必要があります。

次のコードスニペットは、デバイス上でメッセージを作成するとき、それに iothub-creation-time-utc プロパティを追加する方法を示しています。

重要

このタイムスタンプの形式は、タイムゾーン情報を含まない UTC であることが必要です。たとえば、2021-04-21T11:30:16Z は有効で、2021-04-21T11:30:16-07:00 は無効です。

async function sendTelemetry(deviceClient, index) {
  console.log('Sending telemetry message %d...', index);
  const msg = new Message(
    JSON.stringify(
      deviceTemperatureSensor.updateSensor().getCurrentTemperatureObject()
    )
  );
  msg.properties.add("iothub-creation-time-utc", new Date().toISOString());
  msg.contentType = 'application/json';
  msg.contentEncoding = 'utf-8';
  await deviceClient.sendEvent(msg);
}

private static void sendTemperatureTelemetry() {
  String telemetryName = "temperature";
  String telemetryPayload = String.format("{\"%s\": %f}", telemetryName, temperature);

  Message message = new Message(telemetryPayload);
  message.setContentEncoding(StandardCharsets.UTF_8.name());
  message.setContentTypeFinal("application/json");
  message.setProperty("iothub-creation-time-utc", Instant.now().toString());

  deviceClient.sendEventAsync(message, new MessageIotHubEventCallback(), message);
  log.debug("My Telemetry: Sent - {\"{}\": {}°C} with message Id {}.", telemetryName, temperature, message.getMessageId());
  temperatureReadings.put(new Date(), temperature);
}

private async Task SendTemperatureTelemetryAsync()
{
  const string telemetryName = "temperature";

  string telemetryPayload = $"{{ \"{telemetryName}\": {_temperature} }}";
  using var message = new Message(Encoding.UTF8.GetBytes(telemetryPayload))
  {
      ContentEncoding = "utf-8",
      ContentType = "application/json",
  };
  message.Properties.Add("iothub-creation-time-utc", DateTime.UtcNow.ToString("yyyy-MM-ddTHH:mm:ssZ"));
  await _deviceClient.SendEventAsync(message);
  _logger.LogDebug($"Telemetry: Sent - {{ \"{telemetryName}\": {_temperature}°C }}.");
}

async def send_telemetry_from_thermostat(device_client, telemetry_msg):
    msg = Message(json.dumps(telemetry_msg))
    msg.custom_properties["iothub-creation-time-utc"] = datetime.now(timezone.utc).isoformat()
    msg.content_encoding = "utf-8"
    msg.content_type = "application/json"
    print("Sent message")
    await device_client.send_message(msg)

プロパティ変更の形式

各メッセージまたはレコードは、デバイスとクラウドのプロパティの変更を表します。エクスポートされたメッセージに含まれる情報は次のとおりです。

applicationId: IoT Central アプリケーションの ID。
messageSource: メッセージのソース - properties。
messageType: cloudPropertyChange、devicePropertyDesiredChange、または devicePropertyReportedChange のいずれか。
deviceId: テレメトリメッセージを送信したデバイスの ID。
schema: ペイロードスキーマの名前とバージョン。
enqueuedTime: IoT Central がこの変更を検出した時刻。
templateId: デバイスに割り当てられているデバイステンプレートの ID。
properties: 変更されたプロパティの名前と値を含む、変更されたプロパティの配列。プロパティがコンポーネントまたは IoT Edge モジュール内でモデル化されている場合、コンポーネントとモジュールの情報が含まれます。
enrichments: エクスポートに設定されたエンリッチメント。

デバイス接続性の変更形式

各メッセージまたはレコードは、1 台のデバイスからの接続イベントを表します。エクスポートされたメッセージに含まれる情報は次のとおりです。

applicationId: IoT Central アプリケーションの ID。
messageSource: メッセージのソース - deviceConnectivity。
messageType: connected または disconnected のいずれか。
deviceId: 変更されたデバイスの ID。
schema: ペイロードスキーマの名前とバージョン。
templateId: デバイスに割り当てられているデバイステンプレートの ID。
enqueuedTime: IoT Central でこの変更が発生した時刻。
enrichments: エクスポートに設定されたエンリッチメント。

デバイスのライフサイクル変更の形式

各メッセージまたはレコードは、1 つのデバイスに対する 1 つの変更を表します。エクスポートされたメッセージに含まれる情報は次のとおりです。

applicationId: IoT Central アプリケーションの ID。
messageSource: メッセージのソース - deviceLifecycle。
messageType: 発生したエラーの種類。 registered、deleted、provisioned、enabled、disabled、displayNameChanged および deviceTemplateChanged のいずれか。
deviceId: 変更されたデバイスの ID。
schema: ペイロードスキーマの名前とバージョン。
templateId: デバイスに割り当てられているデバイステンプレートの ID。
enqueuedTime: IoT Central でこの変更が発生した時刻。
enrichments: エクスポートに設定されたエンリッチメント。

デバイステンプレートのライフサイクル変更の形式

各メッセージまたはレコードは、1 つの発行されたデバイステンプレートに対する 1 つの変更を表します。エクスポートされたメッセージに含まれる情報は次のとおりです。

applicationId: IoT Central アプリケーションの ID。
messageSource: メッセージのソース - deviceTemplateLifecycle。
messageType: created、updated、または deleted のいずれか。
schema: ペイロードスキーマの名前とバージョン。
templateId: デバイスに割り当てられているデバイステンプレートの ID。
enqueuedTime: IoT Central でこの変更が発生した時刻。
enrichments: エクスポートに設定されたエンリッチメント。

監査ログの形式

各監査ログメッセージでは、IoT Central アプリケーション内の監査可能なエンティティに対してユーザーが開始した変更を表します。エクスポートされたメッセージに含まれる情報は次のとおりです。

actor: エンティティを変更したユーザーに関する情報。
applicationId: IoT Central アプリケーションの ID。
messageSource: メッセージのソース - audit。
messageType: 発生したエラーの種類。 updated、created、deleted のいずれかです。
updated: messageType が updated の場合にのみ存在します。更新の詳細情報を提供します。
resource: 変更されたエンティティの詳細。
schema: ペイロードスキーマの名前とバージョン。
deviceId: 変更されたデバイスの ID。
enqueuedTime: IoT Central でこの変更が発生した時刻。
enrichments: エクスポートに設定されたエンリッチメント。

次のステップ

Azure Data Explorer にエクスポートする方法を理解したので、次のステップとして Webhook へのエクスポートを理解することをお勧めします。