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

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

   

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

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

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

   注意

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

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

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

   

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

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

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

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

6. Data 接続で、次のフィールドに入力し、次へ: スキーマを選択します。

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

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

   

8. データ形式を選択します。 CSV 形式のデータの場合、 ファイルの 見出し行を無視するには、最初のレコードを無視します。 JSON 形式のデータの場合は、[ データ形式のエラーを無視 する] を選択して JSON 形式でデータを取り込むか、multijson 形式でデータを取り込むための選択を解除します。 ネストレベルを選択してテーブル列データの分割方法を決定します。

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

   • データを破棄して新しいデータを取得する: 表示されたデータを破棄し、新しいイベントを検索します。
   • より多くのデータをフェッチする: 既に見つかったイベントに加えて、さらに多くのイベントを検索します。

   注意

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

10. [ 次へ: 概要] を選択します。

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

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);
   

   設定	推奨値	フィールドの説明
   テナントID	xxxxxxxx-xxxxx-xxxx-xxxx-xxxxxxxxx	テナント識別子。 ディレクトリ ID とも呼ばれます。
   サブスクリプションID	xxxxxxxx-xxxxx-xxxx-xxxx-xxxxxxxxx	リソースの作成に使用するサブスクリプション ID。
   clientId	xxxxxxxx-xxxxx-xxxx-xxxx-xxxxxxxxx	ご利用のテナント内のリソースにアクセスできるアプリケーションのクライアント ID。
   クライアントシークレット	PlaceholderClientSecret	ご利用のテナント内のリソースにアクセスできるアプリケーションのクライアント シークレット。
   リソースグループ名	testrg	ご利用のクラスターを含むリソース グループの名前。
   クラスター名	mykustocluster	ご利用のクラスターの名前。
   データベース名	mykustodatabase	ご利用のクラスター内のターゲット データベースの名前。
   eventHubConnectionName (イベントハブ接続名)	myeventhubconnect	データ接続の任意の名前。
   テーブル名	StormEvents	ターゲット データベース内のターゲット テーブルの名前。
   マッピング規則名	StormEvents_CSV_Mapping	ターゲット テーブルに関連付けられている列マッピングの名前。
   データフォーマット	csv	メッセージのデータ形式。
   イベントハブリソースID	リソース ID	インジェスト用のデータを保持しているイベント ハブのリソース ID。
   消費者グループ	$Default	ご利用のイベント ハブのコンシューマー グループ。
   場所	米国中部	データ接続リソースの場所。
   圧縮	Gzip または None	データ圧縮の種類。
   データベースルーティング	複数 または 単一	接続のデータベース ルーティング。 値を Single に設定すると、 databaseName 設定で指定されたクラスター内の単一データベースにデータ接続がルーティングされます。 値を Multi に設定した場合は、 データベースインジェスト プロパティを使用して既定のターゲット データベースをオーバーライドできます。 詳細については、「 イベントルーティング」を参照してください。

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())
   

   設定	推奨値	フィールドの説明
   テナントID	xxxxxxxx-xxxxx-xxxx-xxxx-xxxxxxxxx	テナントID。 ディレクトリ ID とも呼ばれます。
   サブスクリプションID	xxxxxxxx-xxxxx-xxxx-xxxx-xxxxxxxxx	リソースの作成に使用するサブスクリプション ID。
   クライアントID	xxxxxxxx-xxxxx-xxxx-xxxx-xxxxxxxxx	ご利用のテナント内のリソースにアクセスできるアプリケーションのクライアント ID。
   クライアントシークレット	xxxxxxxxxxxxxx	ご利用のテナント内のリソースにアクセスできるアプリケーションのクライアント シークレット。
   リソースグループ名	myresourcegroup	ご利用のクラスターを含むリソース グループの名前。
   クラスター名	mycluster	ご利用のクラスターの名前。
   データベース名 (database_name)	mydatabase	ご利用のクラスター内のターゲット データベースの名前。
   データ接続名	myeventhubconnect	データ接続の任意の名前。
   テーブル名	mytable	ターゲット データベース内のターゲット テーブルの名前。
   マッピングルール名	mytablemappingrule	ターゲット テーブルに関連付けられている列マッピングの名前。
   データ形式	csv	メッセージのデータ形式。
   event_hub_resource_id	リソース ID	インジェスト用のデータを保持しているイベント ハブのリソース ID。
   消費者グループ	$Default	ご利用のイベント ハブのコンシューマー グループ。
   場所	米国中部	データ接続リソースの場所。
   データベースルーティング	複数 または 単一	接続のデータベース ルーティング。 値を Single に設定すると、 databaseName 設定で指定されたクラスター内の単一データベースにデータ接続がルーティングされます。 値を Multi に設定した場合は、 データベースインジェスト プロパティを使用して既定のターゲット データベースをオーバーライドできます。 詳細については、「 イベントルーティング」を参照してください。

次の例は、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. 上部のメニュー バーから [削除] を選択 します。

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

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

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

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

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)