SDK を使用して Azure Data Explorer の Event Grid データ接続を作成する

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

Azure portal または ARM テンプレートを使用して接続を作成する方法については、Event Grid データ接続の作成に関するページを参照してください。

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

Note

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

以前の SDK バージョンに基づくサンプル コードについては、アーカイブ記事を参照してください。

前提条件

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

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

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

C#

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

2. 認証に使用する Microsoft Entra アプリケーション プリンシパルを作成します。 ディレクトリ (テナント) 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 credentials = new ClientSecretCredential(tenantId, clientId, clientSecret);
   var resourceManagementClient = new ArmClient(credentials, subscriptionId);
   var resourceGroupName = "testrg";
   //The cluster and database that are created as part of the Prerequisites
   var clusterName = "mykustocluster";
   var databaseName = "mykustodatabase";
   var subscription = await resourceManagementClient.GetDefaultSubscriptionAsync();
   var resourceGroup = (await subscription.GetResourceGroupAsync(resourceGroupName)).Value;
   var cluster = (await resourceGroup.GetKustoClusterAsync(clusterName)).Value;
   var database = (await cluster.GetKustoDatabaseAsync(databaseName)).Value;
   var dataConnections = database.GetKustoDataConnections();
   var eventGridConnectionName = "myeventgridconnect";
   //The event hub and storage account that are created as part of the Prerequisites
   var eventHubResourceId = new ResourceIdentifier("/subscriptions/<storageAccountSubscriptionId>/resourceGroups/<storageAccountResourceGroupName>/providers/Microsoft.Storage/storageAccounts/<storageAccountName>");
   var storageAccountResourceId = new ResourceIdentifier("/subscriptions/<eventHubSubscriptionId>/resourceGroups/<eventHubResourceGroupName>/providers/Microsoft.EventHub/namespaces/<eventHubNamespaceName>/eventhubs/<eventHubName>");
   var consumerGroup = "$Default";
   var location = AzureLocation.CentralUS;
   //The table and column mapping are created as part of the Prerequisites
   var tableName = "StormEvents";
   var mappingRuleName = "StormEvents_CSV_Mapping";
   var dataFormat = KustoEventGridDataFormat.Csv;
   var blobStorageEventType = BlobStorageEventType.MicrosoftStorageBlobCreated;
   var databaseRouting = KustoDatabaseRouting.Multi;
   var eventGridConnectionData = new KustoEventGridDataConnection
   {
       StorageAccountResourceId = storageAccountResourceId, EventHubResourceId = eventHubResourceId,
       ConsumerGroup = consumerGroup, TableName = tableName, Location = location, MappingRuleName = mappingRuleName,
       DataFormat = dataFormat, BlobStorageEventType = blobStorageEventType, DatabaseRouting = databaseRouting
   };
   await dataConnections.CreateOrUpdateAsync(WaitUntil.Completed, 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 インジェスト プロパティを使用して既定のターゲット データベースをオーバーライドできます。 詳細については、「イベント ルーティング」を参照してください。

Python

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

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

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

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

   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 データ接続を使用する

このセクションでは、BLOB の作成または BLOB の名前変更の後に、Azure Blob Storage または Azure Data Lake Gen 2 からクラスターへのインジェストをトリガーする方法について説明します。

BLOB のアップロードに使用するストレージ SDK の種類に基づいて、関連するタブを選択します。

Azure Blob Storage

次のサンプル コードでは、Azure Blob Storage SDK を使用して Azure Blob Storage にファイルをアップロードします。 アップロードによって Event Grid データ接続がトリガーされ、データが Azure Data Explorer に取り込まれます。

var azureStorageAccountConnectionString=<storage_account_connection_string>;
var containerName = <container_name>;
var blobName = <blob_name>;
var localFileName = <file_to_upload>;
var uncompressedSizeInBytes = <uncompressed_size_in_bytes>;
var mapping = <mappingReference>;
// Create a new container in your storage account.
var azureStorageAccount = CloudStorageAccount.Parse(azureStorageAccountConnectionString);
var blobClient = azureStorageAccount.CreateCloudBlobClient();
var container = blobClient.GetContainerReference(containerName);
container.CreateIfNotExists();
// Set metadata and upload a file to the blob.
var blob = container.GetBlockBlobReference(blobName);
blob.Metadata.Add("rawSizeBytes", uncompressedSizeInBytes);
blob.Metadata.Add("kustoIngestionMappingReference", mapping);
blob.UploadFromFile(localFileName);
// Confirm success of the upload by listing the blobs in your container.
var blobs = container.ListBlobs();

Note

Azure Data Explorer では、BLOB 投稿の取り込みは削除されません。 BLOB の削除を管理する Azure Blob Storage のライフサイクルを使用して、BLOB を 3 から 5 日間保持します。

Azure Data Lake

次のサンプル コードでは、Azure Data Lake SDK を使用して、Data Lake Storage Gen2 にファイルをアップロードします。 アップロードによって Event Grid データ接続がトリガーされ、データが Azure Data Explorer に取り込まれます。

var accountName = <storage_account_name>;
var accountKey = <storage_account_key>;
var fileSystemName = <file_system_name>;
var fileName = <file_name>;
var localFileName = <file_to_upload>;
var uncompressedSizeInBytes = <uncompressed_size_in_bytes>;
var mapping = <mapping_reference>;
var sharedKeyCredential = new StorageSharedKeyCredential(accountName, accountKey);
var dfsUri = "https://" + accountName + ".dfs.core.windows.net";
var dataLakeServiceClient = new DataLakeServiceClient(new Uri(dfsUri), sharedKeyCredential);
// Create the filesystem.
var dataLakeFileSystemClient = dataLakeServiceClient.CreateFileSystem(fileSystemName).Value;
// Define file metadata and uploading options.
IDictionary<String, String> metadata = new Dictionary<string, string>();
metadata.Add("rawSizeBytes", uncompressedSizeInBytes);
metadata.Add("kustoIngestionMappingReference", mapping);
var uploadOptions = new DataLakeFileUploadOptions
{
    Metadata = metadata,
    Close = true // Note: The close option triggers the event being processed by the data connection.
};
// Write to the file.
var dataLakeFileClient = dataLakeFileSystemClient.GetFileClient(fileName);
dataLakeFileClient.Upload(localFileName, uploadOptions);

Note

• Azure Data Lake SDK を使用してファイルをアップロードする場合、初期ファイル作成イベントのサイズは 0 です。これは、データ インジェスト中に Azure Data Explorer によって無視されます。 適切なインジェストを確保するには、Close パラメーターを true に設定します。 このパラメーターを指定すると、upload メソッドにより FlushAndClose イベントがトリガーされます。これは、最終更新が行われ、ファイル ストリームが閉じられたことを示します。
• Event Grid からのトラフィックと、Azure Data Explorer にイベントを取り込む際の後続の処理を減らすには、data.api キーをフィルター処理して FlushAndClose イベントのみを含めることで、サイズ 0 のファイル作成イベントを削除することをお勧めします。 フラッシュについて詳しくは、Azure Data Lake の Flush メソッドに関するページを参照してください。

BLOB を名前変更する

ADLSv2 では、ディレクトリの名前を変更できます。 ただし、ディレクトリの名前を変更しても、BLOB の名前が変更されたイベントがトリガーされたり、ディレクトリ内に含まれる BLOB のインジェストが開始されたりしないことに留意してください。 ディレクトリの名前を変更した後に BLOB の取り込みを確実に行うには、ディレクトリ内の個々の BLOB の名前を直接変更する必要があります。

次のサンプル コードは、ADLSv2 ストレージ アカウント内の BLOB の名前を変更する方法を示しています。

var accountName = <storage_account_name>;
var accountKey = <storage_account_key>;
var fileSystemName = <file_system_name>;
var sourceFilePath = <source_file_path>;
var destinationFilePath = <destination_file_path>;
var sharedKeyCredential = new StorageSharedKeyCredential(accountName, accountKey);
var dfsUri = "https://" + accountName + ".dfs.core.windows.net";
var dataLakeServiceClient = new DataLakeServiceClient(new Uri(dfsUri), sharedKeyCredential);
// Get a client to the the filesystem
var dataLakeFileSystemClient = dataLakeServiceClient.GetFileSystemClient(fileSystemName);
// Rename a file in the file system
var dataLakeFileClient = dataLakeFileSystemClient.GetFileClient(sourceFilePath);
dataLakeFileClient.Rename(destinationFilePath);

Note

データ接続を作成するとき、または Event Grid リソースを手動で作成するときに、特定のサブジェクトを追跡するフィルターを定義した場合、これらのフィルターは対象のファイル パスに適用されます。

Note

階層型名前空間の機能が有効になっているストレージ アカウントでは、CopyBlob 操作に続きインジェストをトリガーすることはサポートされていません。

重要

カスタム コードからストレージ イベントを生成し、Event Hubs に送信することは、極力避けてください。 そうする場合は、生成されたイベントが適切なストレージ イベント スキーマと JSON 形式の仕様に厳密に準拠していることを確認してください。

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

C#

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

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

Python

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