Azure Data Explorer の IoT Hub データ接続を作成する

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

この記事では、ビッグデータのストリーミングプラットフォームとなる IoT インジェストサービスである IoT Hub から Azure Data Explorer にデータを取り込む方法について説明します。

IoT Hub から Azure Data Explorer への取り込みに関する一般的な情報については、IoT Hub への接続に関する記事を参照してください。

Note

データ接続の作成後にエンキューされたイベントのみが取り込まれたます。

前提条件

Azure サブスクリプション。無料の Azure アカウントを作成します。
Azure Data Explorer クラスターとデータベース。クラスターとデータベースを作成します。
宛先テーブル。テーブルを作成するか、既存のテーブルを使用します。
テーブルのインジェストマッピング。
取り込み用のデータが含まれる IoT Hub。

IoT Hub データ接続を作成する

このセクションでは、IoT Hub と Azure Data Explorer テーブルの間の接続を確立します。この接続が確立されている限り、データは IoT Hub からターゲットテーブルに送信されます。

Azure Data Explorer クラスターの左側のメニューで、 Databases を選択し ターゲットテーブルを含むデータベースを選択します。
[データ接続] 、 [データ接続の追加] の順に選択します。ドロップダウンから IoT Hub を選択します。

フォームに次の情報を入力します。

設定	フィールドの説明
データ接続名	Azure Data Explorer で作成する接続の名前
サブスクリプション	Event Hubs リソースが配置されているサブスクリプション ID。
IoT Hub	IoT Hub 名
共有アクセスポリシー	共有アクセスポリシーの名前。読み取りアクセス許可が必要
コンシューマーグループ	IoT Hub の組み込みのエンドポイントに定義されているコンシューマーグループ
イベントシステムプロパティ	IoT Hub イベントのシステムプロパティ。システムプロパティを追加する場合は、テーブルスキーマとマッピングを作成または更新して、選択したプロパティを含めます。

データ接続フォームを示す Azure Data Explorer Web UI のスクリーンショット。

Note

イベントシステムプロパティは、単一レコードイベントでサポートされています。
CSV マッピングの場合、プロパティはレコードの先頭に追加されます。 JSON マッピングの場合、プロパティはドロップダウンリストに表示される名前に従って追加されます。

ユースケースによっては、マルチデータベースルーティングを有効にすることもできます。データベースのルーティングの詳細については、「イベントルーティング」を参照してください。

次のルーティング設定を入力します。

設定	推奨値	フィールドの説明
テーブル名	TestTable	testdb に作成したテーブル。
データ形式	JSON	サポートされている形式は、AVRO、CSV、JSON、ORC、PARQUET、PSV、SCSV、SOHSV、TSV、TXT、TSVE、APACHE AVRO、W3CLOG です。
マッピング	TestMapping	testdb に作成したマッピング。これにより、受信データを testdb の列名とデータ型にマッピングします。指定されていない場合、テーブルのスキーマから派生した ID のデータマッピングが使用されます。

[ターゲットテーブル] フォームの既定のルーティング設定を示す Azure Data Explorer Web UI のスクリーンショット。

Note

JSON データは、既定で multijson として解析されます。 厳密な JSON 形式でデータを取り込むには形式エラーを選択します。
Event システムプロパティを選択した場合は、テーブルスキーマとマッピングにシステムプロパティを含める必要があります。

［作成］ を選択します

警告

手動フェールオーバーの場合データ接続を再作成します。

Microsoft.Azure.Management.Kusto NuGet パッケージをインストールします。
認証に使用する Azure AD アプリケーションプリンシパルを作成します。ディレクトリ (テナント) ID、アプリケーション ID、クライアントシークレットが必要です。

次のコードを実行します。

var tenantId = "xxxxxxxx-xxxxx-xxxx-xxxx-xxxxxxxxx"; //Directory (tenant) ID
var clientId = "xxxxxxxx-xxxxx-xxxx-xxxx-xxxxxxxxx"; //Application ID
var clientSecret = "PlaceholderClientSecret"; //Client Secret
var subscriptionId = "xxxxxxxx-xxxxx-xxxx-xxxx-xxxxxxxxx";
var authClient = ConfidentialClientApplicationBuilder.Create(clientId)
    .WithAuthority($"https://login.microsoftonline.com/{tenantId}")
    .WithClientSecret(clientSecret)
    .Build();
var result = authClient.AcquireTokenForClient(new[] { "https://management.core.windows.net/" }).ExecuteAsync().Result;
var credentials = new TokenCredentials(result.AccessToken, result.TokenType);
var kustoManagementClient = new KustoManagementClient(credentials) { SubscriptionId = subscriptionId };
var resourceGroupName = "testrg";
//The cluster and database that are created as part of the Prerequisites
var clusterName = "mykustocluster";
var databaseName = "mykustodatabase";
var iotHubConnectionName = "myiothubconnect";
//The IoT hub that is created as part of the Prerequisites
var iotHubResourceId = "/subscriptions/<iotHubSubscriptionId>/resourceGroups/<iotHubResourceGroupName>/providers/Microsoft.Devices/IotHubs/<iotHubName>";
var sharedAccessPolicyName = "iothubforread";
var consumerGroup = "$Default";
var location = "Central US";
//The table and column mapping are created as part of the Prerequisites
var tableName = "StormEvents";
var mappingRuleName = "StormEvents_CSV_Mapping";
var dataFormat = DataFormat.CSV;
var databaseRouting = "Multi";
var iotHubConnectionData = new IotHubDataConnection(
    iotHubResourceId, consumerGroup, sharedAccessPolicyName,
    tableName: tableName, location: location, mappingRuleName: mappingRuleName,
    dataFormat: dataFormat, databaseRouting: databaseRouting
);
await kustoManagementClient.DataConnections.CreateOrUpdateAsync(resourceGroupName, clusterName, databaseName, iotHubConnectionName, iotHubConnectionData);

設定	推奨値	フィールドの説明
tenantId	xxxxxxxx-xxxxx-xxxx-xxxx-xxxxxxxxx	テナント ID。ディレクトリ ID とも呼ばれます。
subscriptionId	xxxxxxxx-xxxxx-xxxx-xxxx-xxxxxxxxx	リソースの作成に使用するサブスクリプション ID。
clientId	xxxxxxxx-xxxxx-xxxx-xxxx-xxxxxxxxx	ご利用のテナント内のリソースにアクセスできるアプリケーションのクライアント ID。
clientSecret	PlaceholderClientSecret	ご利用のテナント内のリソースにアクセスできるアプリケーションのクライアントシークレット。
resourceGroupName	testrg	ご利用のクラスターを含むリソースグループの名前。
clusterName	mykustocluster	ご利用のクラスターの名前。
databaseName	mykustodatabase	ご利用のクラスター内のターゲットデータベースの名前。
iotHubConnectionName	myiothubconnect	データ接続の任意の名前。
tableName	StormEvents	ターゲットデータベース内のターゲットテーブルの名前。
mappingRuleName	StormEvents_CSV_Mapping	ターゲットテーブルに関連付けられている列マッピングの名前。
dataFormat	csv	メッセージのデータ形式。
iotHubResourceId	リソース ID	インジェスト用のデータを保持している IoT ハブのリソース ID。
sharedAccessPolicyName	iothubforread	IoT Hub に接続するデバイスとサービスのアクセス許可を定義する共有アクセスポリシーの名前。
consumerGroup	$Default	ご利用のイベントハブのコンシューマーグループ。
location	米国中部	データ接続リソースの場所。
databaseRouting	Multi または Single	接続のデータベースルーティング。値を Single に設定すると、このデータ接続のルーティング先は databaseName 設定で指定されたクラスター内の単一のデータベースとなります。値を Multi に設定すると、Database というインジェストプロパティを使用して既定のターゲットデータベースをオーバーライドできます。詳細については、「イベントルーティング」を参照してください。

必要なライブラリをインストールします。
```
pip install azure-common
pip install azure-mgmt-kusto
```
認証に使用する Azure AD アプリケーションプリンシパルを作成します。ディレクトリ (テナント) ID、アプリケーション ID、クライアントシークレットが必要です。

次のコードを実行します。

from azure.mgmt.kusto import KustoManagementClient
from azure.mgmt.kusto.models import IotHubDataConnection
from azure.common.credentials import ServicePrincipalCredentials

#Directory (tenant) ID
tenant_id = "xxxxxxxx-xxxxx-xxxx-xxxx-xxxxxxxxx"
#Application ID
client_id = "xxxxxxxx-xxxxx-xxxx-xxxx-xxxxxxxxx"
#Client Secret
client_secret = "xxxxxxxxxxxxxx"
subscription_id = "xxxxxxxx-xxxxx-xxxx-xxxx-xxxxxxxxx"
credentials = ServicePrincipalCredentials(
        client_id=client_id,
        secret=client_secret,
        tenant=tenant_id
    )
kusto_management_client = KustoManagementClient(credentials, subscription_id)

resource_group_name = "testrg"
#The cluster and database that are created as part of the Prerequisites
cluster_name = "mykustocluster"
database_name = "mykustodatabase"
data_connection_name = "myeventhubconnect"
#The IoT hub that is created as part of the Prerequisites
iot_hub_resource_id = "/subscriptions/xxxxxxxx-xxxxx-xxxx-xxxx-xxxxxxxxx/resourceGroups/xxxxxx/providers/Microsoft.Devices/IotHubs/xxxxxx";
shared_access_policy_name = "iothubforread"
consumer_group = "$Default"
location = "Central US"
#The table and column mapping that are created as part of the Prerequisites
table_name = "StormEvents"
mapping_rule_name = "StormEvents_CSV_Mapping"
data_format = "csv"
database_routing = "Multi"

#Returns an instance of LROPoller, check https://learn.microsoft.com/python/api/msrest/msrest.polling.lropoller?view=azure-python
poller = kusto_management_client.data_connections.create_or_update(resource_group_name=resource_group_name, cluster_name=cluster_name, database_name=database_name, data_connection_name=data_connection_name,
                                            parameters=IotHubDataConnection(iot_hub_resource_id=iot_hub_resource_id, shared_access_policy_name=shared_access_policy_name,
                                                                                consumer_group=consumer_group, table_name=table_name, location=location, mapping_rule_name=mapping_rule_name, data_format=data_format, database_routing=database_routing))

設定	推奨値	フィールドの説明
tenant_id	xxxxxxxx-xxxxx-xxxx-xxxx-xxxxxxxxx	テナント ID。ディレクトリ ID とも呼ばれます。
subscriptionId	xxxxxxxx-xxxxx-xxxx-xxxx-xxxxxxxxx	リソースの作成に使用するサブスクリプション ID。
client_id	xxxxxxxx-xxxxx-xxxx-xxxx-xxxxxxxxx	ご利用のテナント内のリソースにアクセスできるアプリケーションのクライアント ID。
client_secret	xxxxxxxxxxxxxx	ご利用のテナント内のリソースにアクセスできるアプリケーションのクライアントシークレット。
resource_group_name	testrg	ご利用のクラスターを含むリソースグループの名前。
cluster_name	mykustocluster	ご利用のクラスターの名前。
database_name	mykustodatabase	ご利用のクラスター内のターゲットデータベースの名前。
data_connection_name	myeventhubconnect	データ接続の任意の名前。
table_name	StormEvents	ターゲットデータベース内のターゲットテーブルの名前。
mapping_rule_name	StormEvents_CSV_Mapping	ターゲットテーブルに関連付けられている列マッピングの名前。
data_format	csv	メッセージのデータ形式。
iot_hub_resource_id	リソース ID	インジェスト用のデータを保持している IoT ハブのリソース ID。
shared_access_policy_name	iothubforread	IoT Hub に接続するデバイスとサービスのアクセス許可を定義する共有アクセスポリシーの名前。
consumer_group	$Default	ご利用のイベントハブのコンシューマーグループ。
location	米国中部	データ接続リソースの場所。
database_routing	Multi または Single	接続のデータベースルーティング。値を Single に設定すると、このデータ接続のルーティング先は databaseName 設定で指定されたクラスター内の単一のデータベースとなります。値を Multi に設定すると、Database というインジェストプロパティを使用して既定のターゲットデータベースをオーバーライドできます。詳細については、「イベントルーティング」を参照してください。

次の例は、IoT Hub データ接続を追加するための Azure Resource Manager テンプレートを示しています。フォームを使用して、Azure portal でテンプレートの編集とデプロイを実行できます。

{
    "$schema": "https://schema.management.azure.com/schemas/2015-01-01/deploymentTemplate.json#",
    "contentVersion": "1.0.0.0",
    "parameters": {
        "IotHubs_iothubdemo_name": {
            "type": "string",
            "defaultValue": "iothubdemo",
            "metadata": {
                "description": "Specifies the IoT Hub name."
            }
        },
        "iothubpolices_iothubowner_name": {
            "type": "string",
            "defaultValue": "iothubowner",
            "metadata": {
                "description": "Specifies the shared access policy name."
            }
        },
        "consumergroup_default_name": {
            "type": "string",
            "defaultValue": "$Default",
            "metadata": {
                "description": "Specifies the consumer group of the IoT Hub."
            }
        },
        "Clusters_kustocluster_name": {
            "type": "string",
            "defaultValue": "kustocluster",
            "metadata": {
                "description": "Specifies the name of the cluster"
            }
        },
        "databases_kustodb_name": {
            "type": "string",
            "defaultValue": "kustodb",
            "metadata": {
                "description": "Specifies the name of the database"
            }
        },
        "tables_kustotable_name": {
            "type": "string",
            "defaultValue": "kustotable",
            "metadata": {
                "description": "Specifies the name of the table"
            }
        },
        "mapping_kustomapping_name": {
            "type": "string",
            "defaultValue": "kustomapping",
            "metadata": {
                "description": "Specifies the name of the mapping rule"
            }
        },
        "dataformat_type": {
            "type": "string",
            "defaultValue": "csv",
            "metadata": {
                "description": "Specifies the data format"
            }
        },
           "databaseRouting_type": {
            "type": "string",
            "defaultValue": "Single",
            "metadata": {
                "description": "The database routing for the connection. If you set the value to **Single**, the data connection will be routed to a single database in the cluster as specified in the *databaseName* setting. If you set the value to **Multi**, you can override the default target database using the *Database* EventData property."
            }
        },
        "dataconnections_kustodc_name": {
            "type": "string",
            "defaultValue": "kustodc",
            "metadata": {
                "description": "Name of the data connection to create"
            }
        },
        "subscriptionId": {
            "type": "string",
            "defaultValue": "[subscription().subscriptionId]",
            "metadata": {
                "description": "Specifies the subscriptionId of the IoT Hub"
            }
        },
        "resourceGroup": {
            "type": "string",
            "defaultValue": "[resourceGroup().name]",
            "metadata": {
                "description": "Specifies the resourceGroup of the IoT Hub"
            }
        },
        "location": {
            "type": "string",
            "defaultValue": "[resourceGroup().location]",
            "metadata": {
                "description": "Location for all resources."
            }
        }
    },
    "variables": {
    },
    "resources": [{
            "type": "Microsoft.Kusto/Clusters/Databases/DataConnections",
            "apiVersion": "2019-09-07",
            "name": "[concat(parameters('Clusters_kustocluster_name'), '/', parameters('databases_kustodb_name'), '/', parameters('dataconnections_kustodc_name'))]",
            "location": "[parameters('location')]",
            "kind": "IotHub",
            "properties": {
                "iotHubResourceId": "[resourceId(parameters('subscriptionId'), parameters('resourceGroup'), 'Microsoft.Devices/IotHubs', parameters('IotHubs_iothubdemo_name'))]",
                "consumerGroup": "[parameters('consumergroup_default_name')]",
                "sharedAccessPolicyName": "[parameters('iothubpolices_iothubowner_name')]",
                "tableName": "[parameters('tables_kustotable_name')]",
                "mappingRuleName": "[parameters('mapping_kustomapping_name')]",
                "dataFormat": "[parameters('dataformat_type')]",
                 "databaseRouting": "[parameters('databaseRouting_type')]"
            }
        }
    ]
}

IoT Hub データ接続を削除する

Azure portal から IoT Hub 接続を削除するには、次の操作を行います。

クラスターに移動します。左側のメニューから Databases を選択します。次に、ターゲットテーブルを含むデータベースを選択します。
左側のメニューから、[データ接続を選択します。次に、関連する IoT Hub データ接続の横にあるチェックボックスをオンにします。
上部のメニューバーから Delete を選択します。

IoT Hub 接続を削除するには、次のコマンドを実行します。

kustoManagementClient.DataConnections.Delete(resourceGroupName, clusterName, databaseName, dataConnectionName);

IoT Hub 接続を削除するには、次のコマンドを実行します。

kusto_management_client.data_connections.delete(resource_group_name=resource_group_name, cluster_name=kusto_cluster_name, database_name=kusto_database_name, data_connection_name=kusto_data_connection_name)

次のステップ

Azure Data Explorer でデータのクエリを実行する

次の方法で共有

Azure Data Explorer の IoT Hub データ接続を作成する

前提条件

IoT Hub データ接続を作成する

IoT Hub データ接続を削除する

次のステップ

その他のリソース