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

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

この記事では、Event Grid データ接続を使用して、ストレージアカウントから Azure Data Explorer に BLOB を取り込む方法を説明します。 Azure Event Grid サブスクリプションを設定する Event Grid データ接続を作成します。 Event Grid サブスクリプションは、Azure Event Hubs 経由でストレージアカウントから Azure Data Explorer にイベントをルーティングします。

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

Note

Event Grid 接続で最高のパフォーマンスを実現するには、BLOB メタデータを使用して rawSizeBytes インジェストプロパティを設定します。詳細については、「 ingestion プロパティ」を参照してください。

前提条件

Azure サブスクリプション。無料の Azure アカウントを作成します。
Azure Data Explorer クラスターとデータベース。クラスターとデータベースを作成します。
宛先テーブル。テーブルを作成するか、既存のテーブルを使用します。
テーブルのインジェストマッピング。
ストレージアカウント。 Event Grid 通知サブスクリプションは、 BlobStorage、 StorageV2、または Data Lake Storage Gen2 の Azure Storage アカウントに設定できます。

Event Grid のデータ接続を作成する

このセクションでは、Event Grid と Azure Data Explorer テーブル間の接続を確立します。

作成したクラスターの [データベース]>[TestDatabase] を選択します。
[データインジェスト]>[データ接続の追加] を選択します。

[基本] の下で、接続の種類として [BLOB ストレージ] を選択してから、フォームに次の情報を入力します。

BLOB ストレージ接続タイプのオプションを示す [データ接続] - [基本] タブのスクリーンショット。

設定	推奨値	フィールドの説明
データ接続名	test-grid-connection	Azure Data Explorer で作成する接続の名前。
ストレージアカウントのサブスクリプション	サブスクリプション ID	ストレージアカウントが存在するサブスクリプション ID。
ストレージアカウント	gridteststorage1	作成済みのストレージアカウントの名前。
イベントの種類	"作成された BLOB" または "名前変更された BLOB"	インジェストをトリガーするイベントの種類。 "名前変更された BLOB" は、ADLSv2 ストレージに対してのみサポートされています。 BLOB の名前を変更するには、Azure portal で BLOB に移動し、BLOB を右クリックして Rename を選択します。サポートされる種類は、Microsoft.Storage.BlobCreated または Microsoft.Storage.BlobRenamed。
リソースの作成	自動	Azure Data Explorer で Event Grid サブスクリプション、Event Hubs の名前空間、Event Hubs を自動作成するかどうかを定義します。リソースを手動で作成するには、「Event Grid インジェスト用のリソースを手動で作成する」をご覧ください。

特定のサブジェクトを追跡するには、 [Filter settings](フィルターの設定) を選択します。次のように、通知用のフィルターを設定します。
- [Prefix](プレフィックス) フィールドはサブジェクトの "リテラル" プレフィックスです。適用されるパターンは startswith であるため、複数のコンテナー、フォルダー、BLOB を対象にできます。ワイルドカードは使用できません。
  - BLOB コンテナーに対してフィルターを定義するには、フィールドを /blobServices/default/containers/[container prefix] のように設定する "必要があります"。
  - BLOB プレフィックス (または Azure Data Lake Gen2 のフォルダー) に対してフィルターを定義するには、フィールドを /blobServices/default/containers/[container name]/blobs/[folder/blob prefix] のように設定する "必要があります"。
- [Suffix](サフィックス) フィールドは、BLOB の "リテラル" サフィックスです。ワイルドカードは使用できません。
- [Case-Sensitive](大文字と小文字の区別) フィールドは、プレフィックスとサフィックスのフィルターで大文字と小文字が区別されるかどうかを示します
- イベントのフィルター処理の詳細については、Blob Storage のイベントに関するページを参照してください。
[Next:Ingest properties](次へ: 取り込みのプロパティ) を選択します。

フォームに次の情報を入力します。テーブル名とマッピング名では大文字と小文字が区別されます。

ターゲットテーブルのプロパティを示す [データ接続] - [Ingest properties] (取り込みのプロパティ) タブのスクリーンショット。

Ingest properties (取り込みのプロパティ):

設定	推奨値	フィールドの説明
他のデータベースへのデータのルーティングを許可する (マルチデータベースデータ接続)	Don't allow	このオプションは、データ接続に関連付けられている既定のターゲットデータベースをオーバーライドする場合にオンにします。データベースのルーティングの詳細については、「イベントルーティング」を参照してください。
テーブル名	TestTable	TestDatabase に作成したテーブル。
データ形式	JSON	サポートされている形式は、Avro、CSV、JSON、MULTILINE JSON、ORC、PARQUET、PSV、SCSV、SOHSV、TSV、TXT、TSVE、APACHEAVRO、RAW、および W3CLOG です。サポートされている圧縮オプションは、Zip と Gzip です。
マッピング	TestMapping	TestDatabase で作成したマッピング。受信データは、TestTable の列名とデータ型にマップされます。指定されていない場合、テーブルのスキーマから派生した ID のデータマッピングが使用されます。
詳細設定	データにヘッダーが含まれている	ヘッダーを無視します。 *SV 型ファイルでサポートされています。

注意

既定のルーティング設定をすべて指定する必要はありません。部分的な設定も受け入れられます。

[Next: Review + create]$次へ: 確認と作成$ を選択します。
*[確認と作成] が表示されます。
自動的に作成されたリソースを確認し、 [作成] を選択します。
デプロイが完了するまでお待ちください。デプロイが失敗した場合は、失敗したステージの横にある [操作の詳細] を選択して、失敗の原因に関する詳細情報を取得します。リソースのデプロイを再試行するには、 [再デプロイ] を選択します。デプロイの前にパラメーターを変更できます。

Azure portal でストレージアカウントに移動します。左側のメニューから [イベント] を選択します
メインウィンドウで、[Azure Data Explorer] タブを選択します。

[データ接続] ペインが開き、[基本] タブが選択されます。

[基本] の下で、フォームに次の情報を入力します。

接続タイプのオプションを示す [データ接続] - [基本] タブのスクリーンショット。

設定	推奨値	フィールドの説明
データ接続名	test-grid-connection	Azure Data Explorer で作成する接続の名前。
ストレージアカウント	gridteststorage1	このウィザードへのアクセスに使用したストレージアカウント。自動で入力されます。
イベントの種類	"作成された BLOB" または "名前変更された BLOB"	インジェストをトリガーするイベントの種類。 "名前変更された BLOB" は、ADLSv2 ストレージに対してのみサポートされています。 BLOB の名前を変更するには、Azure portal で BLOB に移動し、BLOB を右クリックして Rename を選択します。サポートされる種類は、Microsoft.Storage.BlobCreated または Microsoft.Storage.BlobRenamed。
リソースの作成	自動	Azure Data Explorer で Event Grid サブスクリプション、Event Hubs の名前空間、Event Hubs を自動作成するかどうかを定義します。リソースを手動で作成するには、「Event Grid インジェスト用のリソースを手動で作成する」をご覧ください。

[次へ:> 取り込みのプロパティ] を選択します。
[取り込みのプロパティ] が表示されます。

フォームに次の情報を入力します。テーブル名とマッピング名では大文字と小文字が区別されます。

サブスクリプションとデータルーティングのプロパティを示す [データ接続] - [Ingest properties] (取り込みのプロパティ) タブのスクリーンショット。

設定	推奨値	フィールドの説明
サブスクリプション		対象の Azure Data Explorer サブスクリプション。
クラスター名	TestCluster	データの取り込み先となるクラスターの名前。
データベース名	TestDatabase	TestCluster 内に作成したターゲットデータベース。
他のデータベースへのデータのルーティングを許可する (マルチデータベースデータ接続)	Don't allow	このオプションは、データ接続に関連付けられている既定のターゲットデータベースをオーバーライドする場合にオンにします。データベースのルーティングの詳細については、「イベントルーティング」を参照してください。
テーブル名	TestTable	TestDatabase 内に作成したターゲットテーブル。
データ形式	JSON	サポートされている形式は、Avro、CSV、JSON、MULTILINE JSON、ORC、PARQUET、PSV、SCSV、SOHSV、TSV、TXT、TSVE、APACHEAVRO、RAW、および W3CLOG です。サポートされている圧縮オプションは、Zip と Gzip です。
マッピング	TestMapping	TestDatabase で作成したマッピング。受信データは、TestTable の列名とデータ型にマップされます。指定されていない場合、テーブルのスキーマから派生した ID のデータマッピングが使用されます。
詳細設定	データにヘッダーが含まれている	ヘッダーを無視します。 *SV 型ファイルでサポートされています。

Note

データのルーティング設定はすべて指定する必要はありません。部分的な設定も受け入れられます。

[Next: Review + create]$次へ: 確認と作成$ を選択します。
[確認と作成] が表示されます。
自動的に作成されたリソースを確認し、 [作成] を選択します。
デプロイが完了するまでお待ちください。デプロイが失敗した場合は、失敗したステージの横にある [操作の詳細] を選択して、失敗の原因に関する詳細情報を取得します。リソースのデプロイを再試行するには、 [再デプロイ] を選択します。デプロイの前にパラメーターを変更できます。

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 eventGridConnectionName = "myeventgridconnect";
//The event hub and storage account that are created as part of the Prerequisites
var eventHubResourceId = "/subscriptions/<storageAccountSubscriptionId>/resourceGroups/<storageAccountResourceGroupName>/providers/Microsoft.Storage/storageAccounts/<storageAccountName>";
var storageAccountResourceId = "/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 mappingRuleName = "StormEvents_CSV_Mapping";
var dataFormat = DataFormat.CSV;
var blobStorageEventType = "Microsoft.Storage.BlobCreated";
var databaseRouting = "Multi";
var eventGridConnectionData = new EventGridDataConnection(
    storageAccountResourceId, eventHubResourceId, consumerGroup,
    tableName: tableName, location: location, mappingRuleName: mappingRuleName,
    dataFormat: dataFormat, blobStorageEventType: blobStorageEventType, databaseRouting: databaseRouting
);
await kustoManagementClient.DataConnections.CreateOrUpdateAsync(
    resourceGroupName, clusterName, databaseName, eventGridConnectionName, eventGridConnectionData
);

設定	推奨値	フィールドの説明
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	ご利用のクラスター内のターゲットデータベースの名前。
eventGridConnectionName	myeventgridconnect	データ接続の任意の名前。
tableName	StormEvents	ターゲットデータベース内のターゲットテーブルの名前。
mappingRuleName	StormEvents_CSV_Mapping	ターゲットテーブルに関連付けられている列マッピングの名前。
dataFormat	csv	メッセージのデータ形式。
eventHubResourceId	リソース ID	Event Grid がイベントを送信するように構成されているイベントハブのリソース ID。
storageAccountResourceId	リソース ID	インジェスト用のデータを保持しているストレージアカウントのリソース ID。
consumerGroup	$Default	ご利用のイベントハブのコンシューマーグループ。
location	米国中部	データ接続リソースの場所。
blobStorageEventType	Microsoft.Storage.BlobCreated	インジェストをトリガーするイベントの種類。サポートされているイベントは、Microsoft.Storage.BlobCreated または Microsoft.Storage.BlobRenamed です。 BLOB の名前変更は、ADLSv2 ストレージに対してのみサポートされています。
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 EventGridDataConnection
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 event hub and storage account that are created as part of the Prerequisites
event_hub_resource_id = "/subscriptions/xxxxxxxx-xxxxx-xxxx-xxxx-xxxxxxxxx/resourceGroups/xxxxxx/providers/Microsoft.EventHub/namespaces/xxxxxx/eventhubs/xxxxxx"
storage_account_resource_id = "/subscriptions/xxxxxxxx-xxxxx-xxxx-xxxx-xxxxxxxxx/resourceGroups/xxxxxx/providers/Microsoft.Storage/storageAccounts/xxxxxx"
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"
blob_storage_event_type = "Microsoft.Storage.BlobCreated"

#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.begin_create_or_update(resource_group_name=resource_group_name, cluster_name=cluster_name, database_name=database_name, data_connection_name=data_connection_name,
                                            parameters=EventGridDataConnection(storage_account_resource_id=storage_account_resource_id, event_hub_resource_id=event_hub_resource_id, 
                                                                                consumer_group=consumer_group, table_name=table_name, location=location, mapping_rule_name=mapping_rule_name, data_format=data_format, database_routing=database_routing,
                                                                                blob_storage_event_type=blob_storage_event_type))
# The creation of the connection is async. Validation errors are only visible if you wait for the results.
poller.wait()
print(poller.result())

設定	推奨値	フィールドの説明
tenant_id	xxxxxxxx-xxxxx-xxxx-xxxx-xxxxxxxxx	テナント ID。ディレクトリ ID とも呼ばれます。
subscription_id	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	ターゲットテーブルに関連付けられている列マッピングの名前。
database_routing	Multi または Single	接続のデータベースルーティング。値を Single に設定すると、databaseName 設定で指定したクラスター内の単一のデータベースにデータ接続はルーティングされます。値を Multi に設定すると、Database というインジェストプロパティを使用して既定のターゲットデータベースをオーバーライドできます。詳細については、「イベントルーティング」を参照してください。
data_format	csv	メッセージのデータ形式。
event_hub_resource_id	リソース ID	Event Grid がイベントを送信するように構成されているイベントハブのリソース ID。
storage_account_resource_id	リソース ID	インジェスト用のデータを保持しているストレージアカウントのリソース ID。
consumer_group	$Default	ご利用のイベントハブのコンシューマーグループ。
location	米国中部	データ接続リソースの場所。
blob_storage_event_type	Microsoft.Storage.BlobCreated	インジェストをトリガーするイベントの種類。サポートされているイベントは、Microsoft.Storage.BlobCreated または Microsoft.Storage.BlobRenamed です。 BLOB の名前変更は、ADLSv2 ストレージに対してのみサポートされています。

次の例は、Event Grid データ接続を追加するための 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 hub 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."
            }
        },
        "StorageAccounts_storagedemo_name": {
            "type": "string",
            "defaultValue": "storagedemo",
            "metadata": {
                "description": "Specifies the storage account name"
            }
        },
        "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 resources"
            }
        },
        "resourceGroup": {
            "type": "string",
            "defaultValue": "[resourceGroup().name]",
            "metadata": {
                "description": "Specifies the resourceGroup of the resources"
            }
        },
        "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": "EventGrid",
            "properties": {
                "managedIdentityResourceId": "[resourceId('Microsoft.Kusto/clusters', parameters('clusters_kustocluster_name'))]",
                "storageAccountResourceId": "[resourceId(parameters('subscriptionId'), parameters('resourceGroup'), 'Microsoft.Storage/storageAccounts', parameters('StorageAccounts_storagedemo_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')]"
            }
        }
    ]
}

Event Grid データ接続を削除する

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

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

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

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

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

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 Stream Analytics を使用してイベントハブからのデータを処理する

次の方法で共有

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

前提条件

Event Grid のデータ接続を作成する

Event Grid データ接続を削除する

次のステップ

その他のリソース