Azure Data Explorer の Event Hubs データ接続を作成する

Azure Data Explorer では、ビッグ データ ストリーミング プラットフォームとイベント インジェストのサービスである Event Hubs からのインジェストを行うことができます。 Event Hubs は、1 秒あたり数百万件のイベントをほぼリアルタイムで処理できます。

この記事では、イベント ハブに接続し、Azure Data Explorer にデータを取り込みます。 Event Hubs からの取り込みの概要については、「Azure Event Hubs データ接続」を参照してください。

前提条件

• Azure サブスクリプション。 無料の Azure アカウントを作成します。
• Azure Data Explorer クラスターとデータベース。 クラスターとデータベースを作成します。
• 宛先テーブル。 テーブルを作成するか、既存のテーブルを使用する。
• テーブルのインジェスト マッピング。
• インジェスト対象のデータが存在するイベント ハブ。

イベント ハブ データ接続を作成する

このセクションでは、イベント ハブと Azure Data Explorer テーブルの間に接続を確立します。 この接続が存在する間は、データがイベント ハブからターゲットのテーブルに送信されます。 イベント ハブが別のリソースまたはサブスクリプションに移動された場合は、接続を更新または再作成する必要があります。

ポータル

1. Azure portal でクラスターに移動し、 Databases を選択します。 次に、ターゲット テーブルを含むデータベースを選択します。

   

2. 左側のメニューから、 Data インジェストを選択します。 次に、上部のバーで[データ接続の追加] 選択。

3. フォームに次の情報を入力し、 [作成] を選択します。

   設定	推奨値	フィールドの説明
   データ接続名	test-hub-connection	Azure データ エクスプローラーで作成する接続の名前。
   サブスクリプション		イベントハブ リソースが配置されているサブスクリプション ID。
   イベント ハブの名前空間	一意の名前空間名	以前に選択した、名前空間を識別する名前。
   イベント ハブ	test-hub	作成したイベント ハブ。
   コンシューマー グループ	test-group	作成したイベント ハブに定義されているコンシューマー グループ。
   イベント システム プロパティ	関連するプロパティを選択する	イベント ハブのシステム プロパティ。 イベント メッセージごとに複数のレコードがある場合は、システム プロパティが最初のレコードに追加されます。 システム プロパティを追加する場合は、テーブル スキーマとマッピングを作成または更新して、選択したプロパティを含めます。
   圧縮	なし	イベント ハブ メッセージ ペイロードの圧縮の種類。 サポートされている圧縮の種類は、"なし、Gzip" です。
   マネージド ID (推奨)	システム割り当て	イベント ハブに対する読み取りアクセスのために Data Explorer クラスターによって使用されるマネージド ID。 イベント ハブへのアクセスを制御するために、マネージド ID を使用することをお勧めします。 注: データ接続が作成されると、次のようになります。 * "システム割り当て" ID が存在しない場合は自動的に作成されます * マネージド ID は、"Azure Event Hubs データ受信者" ロールが自動的に割り当てられ、Data Explorer クラスターに追加されます。 ロールが割り当てられ、ID がクラスターに追加されたことを確認することをお勧めします。

   Note

   マネージド ID を使用していない既存のデータ接続がある場合は、マネージド ID を使用するように更新することをお勧めします。

ウィザード

次の手順では、 Azure Data Explorer Web UI のインジェスト ウィザードを使用してイベント ハブ接続を作成する手順について説明します。

Note

• ウィザードを使用するには、少なくとも Database ユーザー アクセス許可が必要です。
• パブリック アクセスなしでクラスターとストレージ アカウント間のアクセスを有効にするには、「 マネージド プライベート エンドポイントの作成を参照してください。
• クラスターとイベント ハブは、同じテナントに関連付ける必要があります。 そうでない場合は、C# や Python などの SDK オプションのいずれかを使用します。

1. Azure Data Explorer Web UI の [Data] タブでイベント ハブの Ingest データ カードから Ingest を選択します。

   

2. Ingest data ウィンドウが開き、[Destination] タブが選択されています。 Cluster フィールドと Database フィールドが自動的に設定されます。 ドロップダウン メニューから別のクラスターまたはデータベースを選択できます。

3. [ テーブル]で[新しいテーブル 選択し 新しいテーブルの名前を入力します。 または、既存のテーブルを使用します。

4. [Next: Source](次へ: ソース) を選択します。

5. [ソースの種類] で、 [イベント ハブ] を選択します。

6. [ Data 接続で、次のフィールドに入力し、 Next: Schema を選択します。

   設定	推奨値	フィールドの説明
   サブスクリプション		イベントハブ リソースが配置されているサブスクリプション ID。
   イベント ハブの名前空間		名前空間を識別する名前。
   イベント ハブ		使用するイベント ハブ。
   データ接続名	TestDataConnection	データ接続を識別する名前。
   コンシューマー グループ		イベント ハブに定義されているコンシューマー グループ。
   Compression		イベント ハブ メッセージ ペイロードの圧縮の種類。
   イベント システム プロパティ	関連するプロパティを選択する	イベント ハブのシステム プロパティ。 イベント メッセージごとに複数のレコードがある場合、システム プロパティは最初のものに追加されます。 システム プロパティを追加する場合は、テーブル スキーマとマッピングを作成または更新して、選択したプロパティを含めます。
   イベント取得の開始日	協定世界時 (UTC)	データ接続は、 Event 取得の開始日後に作成された既存の Event Hubs イベントを取得します。 取得できるのは、Event Hubs のリテンション期間によって保持されるイベントのみです。 Event 取得の開始日が指定されていない場合、既定の時刻はデータ接続が作成される時刻です。

7. インジェスト ポリシーを設定します。 クラスターに対して streaming が有効になっている場合は、 ストリーミング インジェストを選択できます。 クラスターでストリーミングが有効になっていない場合は、 Batching 時間を設定します。 Event Hubs の場合、推奨される バッチ処理時間 は 30 秒です。

   

8. Data 形式を選択します。 CSV 形式のデータの場合 ファイルの見出し行を無視するには 最初のレコードを配置します。 JSON 形式のデータの場合は、 [データ形式のエラー を選択して JSON 形式でデータを取り込むか、選択を解除して multijson 形式でデータを取り込みます。 Nested レベルを選択して、テーブル列のデータ除算を決定します。

9. プレビュー ウィンドウに表示されるデータが完全でない場合は、必要なすべてのデータ フィールドを含むテーブルを作成するために、さらにデータが必要になることがあります。 次のコマンドを使用して、イベント ハブから新しいデータをフェッチします。

   • Discard and fetch new data (破棄して新しいデータをフェッチする) : 表示されたデータを破棄し、新しいイベントを検索します。
   • Fetch more data (さらにデータをフェッチする) :既に見つかったイベントに加えて、さらにイベントを検索します。

   注意

   データのプレビューを表示するには、イベント ハブがイベントを送信している必要があります。

10. 概要 を選択します。

11. 確立された Event Hub からの継続的なインジェスト ウィンドウでは、確立が正常に完了すると、すべての手順が緑色のチェック マークでマークされます。

C#

1. Microsoft.Azure.Management.Kusto NuGet パッケージをインストールします。

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

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

   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 eventHubConnectionName = "myeventhubconnect";
   //The event hub that is created as part of the Prerequisites
   var eventHubResourceId = "/subscriptions/<eventHubSubscriptionId>/resourceGroups/<eventHubResourceGroupName>/providers/Microsoft.EventHub/namespaces/<eventHubNamespaceName>/eventhubs/<eventHubName>";
   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 = EventHubDataFormat.CSV;
   var compression = "None";
   var databaseRouting = "Multi";
   var eventHubConnectionData = new EventHubDataConnection(
       eventHubResourceId, consumerGroup,
       location: location, tableName: tableName, mappingRuleName: mappingRuleName,
       dataFormat: dataFormat, compression: compression, databaseRouting: databaseRouting
   );
   await kustoManagementClient.DataConnections.CreateOrUpdateAsync(resourceGroupName, clusterName, databaseName, eventHubConnectionName, eventHubConnectionData);
   

   設定	推奨値	フィールドの説明
   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	ご利用のクラスター内のターゲット データベースの名前。
   eventHubConnectionName	myeventhubconnect	データ接続の任意の名前。
   tableName	StormEvents	ターゲット データベース内のターゲット テーブルの名前。
   mappingRuleName	StormEvents_CSV_Mapping	ターゲット テーブルに関連付けられている列マッピングの名前。
   dataFormat	csv	メッセージのデータ形式。
   eventHubResourceId	リソース ID	インジェスト用のデータを保持しているイベント ハブのリソース ID。
   consumerGroup	$Default	ご利用のイベント ハブのコンシューマー グループ。
   location	米国中部	データ接続リソースの場所。
   compression	Gzip または None	データ圧縮の種類。
   databaseRouting	Multi または Single	接続のデータベース ルーティング。 値を Single に設定すると、このデータ接続のルーティング先は databaseName 設定で指定されたクラスター内の単一のデータベースとなります。 値を Multi に設定すると、Database というインジェスト プロパティを使用して既定のターゲット データベースをオーバーライドできます。 詳細については、「イベント ルーティング」を参照してください。

Python

1. 必要なライブラリをインストールします。

   pip install azure-common
   pip install azure-mgmt-kusto
   

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

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

   from azure.mgmt.kusto import KustoManagementClient
   from azure.mgmt.kusto.models import EventHubDataConnection
   from azure.identity import ClientSecretCredential
   
   #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 = "myresourcegroup"
   #The cluster and database that are created as part of the Prerequisites
   cluster_name = "mycluster"
   database_name = "mydatabase"
   data_connection_name = "myeventhubconnect"
   #The event hub that is created as part of the Prerequisites
   event_hub_resource_id = "/subscriptions/xxxxxxxx-xxxxx-xxxx-xxxx-xxxxxxxxx/resourceGroups/myresourcegroup/providers/Microsoft.EventHub/namespaces/myeventhubnamespace/eventhubs/myeventhub"";
   consumer_group = "$Default"
   location = "Central US"
   #The table and column mapping that are created as part of the Prerequisites
   table_name = "mytable"
   mapping_rule_name = "mytablemappingrule"
   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=EventHubDataConnection(
                   event_hub_resource_id=event_hub_resource_id,
                   consumer_group=consumer_group,
                   location=location,
                   table_name=table_name,
                   mapping_rule_name=mapping_rule_name,
                   data_format=data_format,
                   database_routing=database_routing
               )
           )
   poller.wait()
   print(poller.result())
   

   設定	推奨値	フィールドの説明
   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	myresourcegroup	ご利用のクラスターを含むリソース グループの名前。
   cluster_name	mycluster	ご利用のクラスターの名前。
   database_name	mydatabase	ご利用のクラスター内のターゲット データベースの名前。
   data_connection_name	myeventhubconnect	データ接続の任意の名前。
   table_name	mytable	ターゲット データベース内のターゲット テーブルの名前。
   mapping_rule_name	mytablemappingrule	ターゲット テーブルに関連付けられている列マッピングの名前。
   data_format	csv	メッセージのデータ形式。
   event_hub_resource_id	リソース ID	インジェスト用のデータを保持しているイベント ハブのリソース ID。
   consumer_group	$Default	ご利用のイベント ハブのコンシューマー グループ。
   location	米国中部	データ接続リソースの場所。
   databaseRouting	Multi または Single	接続のデータベース ルーティング。 値を Single に設定すると、このデータ接続のルーティング先は databaseName 設定で指定されたクラスター内の単一のデータベースとなります。 値を Multi に設定すると、Database というインジェスト プロパティを使用して既定のターゲット データベースをオーバーライドできます。 詳細については、「イベント ルーティング」を参照してください。

ARM テンプレート

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

{
    "$schema": "https://schema.management.azure.com/schemas/2015-01-01/deploymentTemplate.json#",
    "contentVersion": "1.0.0.0",
    "parameters": {
        "namespaces_eventhubns_name": {
            "type": "string",
            "defaultValue": "eventhubns",
            "metadata": {
                "description": "Specifies the Event Hubs Namespace name."
            }
        },
        "EventHubs_eventhubdemo_name": {
            "type": "string",
            "defaultValue": "eventhubdemo",
            "metadata": {
                "description": "Specifies the event hub name."
            }
        },
        "consumergroup_default_name": {
            "type": "string",
            "defaultValue": "$Default",
            "metadata": {
                "description": "Specifies the consumer group of the event 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 event hub"
            }
        },
        "resourceGroup": {
            "type": "string",
            "defaultValue": "[resourceGroup().name]",
            "metadata": {
                "description": "Specifies the resourceGroup of the event hub"
            }
        },
        "location": {
            "type": "string",
            "defaultValue": "[resourceGroup().location]",
            "metadata": {
                "description": "Location for all resources."
            }
        }
    },
    "variables": {
    },
    "resources": [{
            "type": "Microsoft.Kusto/Clusters/Databases/DataConnections",
            "apiVersion": "2022-02-01",
            "name": "[concat(parameters('Clusters_kustocluster_name'), '/', parameters('databases_kustodb_name'), '/', parameters('dataconnections_kustodc_name'))]",
            "location": "[parameters('location')]",
            "kind": "EventHub",
            "properties": {
                "managedIdentityResourceId": "[resourceId('Microsoft.Kusto/clusters', parameters('clusters_kustocluster_name'))]",
                "eventHubResourceId": "[resourceId(parameters('subscriptionId'), parameters('resourceGroup'), 'Microsoft.EventHub/namespaces/eventhubs', parameters('namespaces_eventhubns_name'), parameters('EventHubs_eventhubdemo_name'))]",
                "consumerGroup": "[parameters('consumergroup_default_name')]",
                "tableName": "[parameters('tables_kustotable_name')]",
                "mappingRuleName": "[parameters('mapping_kustomapping_name')]",
                "dataFormat": "[parameters('dataformat_type')]",
                "databaseRouting": "[parameters('databaseRouting_type')]"
            }
        }
    ]
}

イベント ハブ データ接続を削除する

ポータル

Azure portal からイベント ハブ接続を削除するには、次の操作を行います。

1. クラスターに移動します。 左側のメニューで [データベース] を選択します。 対象となるテーブルを含むデータベースを選択します。
2. 左側のメニューで [データ接続] を選択します。 次に、関連するイベント ハブ データ接続の横にあるチェック ボックスをオンにします。
3. 上部のメニュー バーで [削除] を選択します。

ウィザード

最後のウィザード ページで、 Manage データ接続 カードを選択します。 ウィザードを既に終了している場合は、前のタブで説明したように、Azure portal を使用してデータ接続を削除します。

C#

イベント ハブ接続を削除するには、次のコマンドを実行します。

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

Python

イベント ハブ接続を削除するには、次のコマンドを実行します。

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 でデータのクエリを実行する