Membuat koneksi data Event Grid untuk Azure Data Explorer dengan SDK

Dalam artikel ini, Anda mempelajari cara menyerap blob dari akun penyimpanan Anda ke Azure Data Explorer menggunakan koneksi data Event Grid. Anda akan membuat koneksi data Event Grid yang menetapkan langganan Azure Event Grid. Langganan Event Grid merutekan peristiwa dari akun penyimpanan Anda ke Azure Data Explorer melalui Azure Event Hubs.

Untuk mempelajari cara membuat koneksi di portal Azure atau dengan templat ARM, lihat Membuat koneksi data Event Grid.

Untuk informasi umum tentang menyerap ke Azure Data Explorer dari Event Grid, lihat Menyambungkan ke Event Grid.

Catatan

Untuk mencapai performa terbaik dengan koneksi Event Grid, atur rawSizeBytes properti penyerapan melalui metadata blob. Untuk informasi selengkapnya, lihat properti penyerapan.

Untuk sampel kode berdasarkan versi SDK sebelumnya, lihat artikel yang diarsipkan.

Prasyarat

• Langganan Azure. Membuat akun Azure gratis.
• Kluster dan database Azure Data Explorer. Membuat kluster dan database.
• Tabel tujuan. Buat tabel atau gunakan tabel yang sudah ada.
• Pemetaan penyerapan untuk tabel.
• Akun penyimpanan. Langganan pemberitahuan Event Grid dapat diatur di akun Azure Storage untuk BlobStorage, , StorageV2atau Data Lake Storage Gen2.
• Daftarkan penyedia sumber daya Event Grid.

Membuat koneksi data Event Grid

Di bagian ini, Anda akan membuat koneksi antara Event Grid dan tabel Azure Data Explorer Anda.

C#

1. Instal paket Microsoft.Azure.Management.Kusto NuGet.

2. Buat perwakilan aplikasi Microsoft Entra untuk digunakan untuk autentikasi. Anda akan memerlukan ID direktori (penyewa), ID aplikasi, dan rahasia klien.

3. Jalankan kode berikut.

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

   Pengaturan	Nilai yang disarankan	Deskripsi bidang
   tenantId	xxxxxxxx-xxxxx-xxxx-xxxx-xxxxxxxxx	ID penyewa Anda. Juga dikenal sebagai ID direktori.
   subscriptionId	xxxxxxxx-xxxxx-xxxx-xxxx-xxxxxxxxx	ID langganan yang Anda gunakan untuk pembuatan sumber daya.
   clientId	xxxxxxxx-xxxxx-xxxx-xxxx-xxxxxxxxx	ID klien aplikasi yang dapat mengakses sumber daya di penyewa Anda.
   clientSecret	PlaceholderClientSecret	Rahasia klien aplikasi yang dapat mengakses sumber daya di penyewa Anda.
   resourceGroupName	testrg	Nama grup sumber daya yang berisi kluster Anda.
   clusterName	mykustocluster	Nama kluster Anda.
   databaseName	mykustodatabase	Nama database target dalam kluster Anda.
   eventGridConnectionName	myeventgridconnect	Nama koneksi data yang diinginkan.
   tableName	StormEvents	Nama tabel target dalam database target.
   mappingRuleName	StormEvents_CSV_Mapping	Nama pemetaan kolom Anda yang terkait dengan tabel target.
   dataFormat	csv	Format data pesan.
   eventHubResourceId	ID Sumber Daya	ID sumber daya hub peristiwa Anda tempat Event Grid dikonfigurasi untuk mengirim peristiwa.
   storageAccountResourceId	ID Sumber Daya	ID sumber daya akun penyimpanan Anda yang menyimpan data untuk penyerapan.
   consumerGroup	$Default	Grup konsumen pusat aktivitas Anda.
   lokasi	US Tengah	Lokasi sumber daya koneksi data.
   blobStorageEventType	Microsoft.Storage.BlobDeleted	Jenis peristiwa yang memicu konsumsi. Peristiwa yang didukung adalah: Microsoft.Storage.BlobCreated atau Microsoft.Storage.BlobRenamed. Penggantian nama blob hanya didukung untuk penyimpanan ADLSv2.
   databaseRouting	Multi atau Tunggal	Perutean database untuk koneksi. Jika Anda mengatur nilai ke Tunggal, koneksi data akan dirutekan ke database tunggal di kluster seperti yang ditentukan dalam pengaturan databaseName . Jika Anda mengatur nilai ke Multi, Anda bisa mengambil alih database target default menggunakan properti Penyerapan database. Untuk informasi selengkapnya, lihat Perutean peristiwa.

Python

1. Instal pustaka yang diperlukan.

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

2. Buat perwakilan aplikasi Microsoft Entra untuk digunakan untuk autentikasi. Anda akan memerlukan ID direktori (penyewa), ID aplikasi, dan rahasia klien.

3. Jalankan kode berikut.

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

   Pengaturan	Nilai yang disarankan	Deskripsi bidang
   tenant_id	xxxxxxxx-xxxxx-xxxx-xxxx-xxxxxxxxx	ID penyewa Anda. Juga dikenal sebagai ID direktori.
   subscription_id	xxxxxxxx-xxxxx-xxxx-xxxx-xxxxxxxxx	ID langganan yang Anda gunakan untuk pembuatan sumber daya.
   client_id	xxxxxxxx-xxxxx-xxxx-xxxx-xxxxxxxxx	ID klien aplikasi yang dapat mengakses sumber daya di penyewa Anda.
   client_secret	xxxxxxxxxxxxxxxx	Rahasia klien aplikasi yang dapat mengakses sumber daya di penyewa Anda.
   resource_group_name	testrg	Nama grup sumber daya yang berisi kluster Anda.
   cluster_name	mykustocluster	Nama kluster Anda.
   database_name	mykustodatabase	Nama database target dalam kluster Anda.
   data_connection_name	myeventhubconnect	Nama koneksi data yang diinginkan.
   table_name	StormEvents	Nama tabel target dalam database target.
   mapping_rule_name	StormEvents_CSV_Mapping	Nama pemetaan kolom Anda yang terkait dengan tabel target.
   database_routing	Multi atau Tunggal	Perutean database untuk koneksi. Jika Anda mengatur nilai ke Tunggal, koneksi data akan dirutekan ke database tunggal di kluster seperti yang ditentukan dalam pengaturan databaseName . Jika Anda mengatur nilai ke Multi, Anda bisa mengambil alih database target default menggunakan properti Penyerapan database. Untuk informasi selengkapnya, lihat Perutean peristiwa.
   data_format	csv	Format data pesan.
   event_hub_resource_id	ID Sumber Daya	ID sumber daya hub peristiwa Anda tempat Event Grid dikonfigurasi untuk mengirim peristiwa.
   storage_account_resource_id	ID Sumber Daya	ID sumber daya akun penyimpanan Anda yang menyimpan data untuk penyerapan.
   consumer_group	$Default	Grup konsumen pusat aktivitas Anda.
   lokasi	US Tengah	Lokasi sumber daya koneksi data.
   blob_storage_event_type	Microsoft.Storage.BlobDeleted	Jenis peristiwa yang memicu konsumsi. Peristiwa yang didukung adalah: Microsoft.Storage.BlobCreated atau Microsoft.Storage.BlobRenamed. Penggantian nama blob hanya didukung untuk penyimpanan ADLSv2.

Menggunakan koneksi data Event Grid

Bagian ini menunjukkan cara memicu penyerapan dari Azure Blob Storage atau Azure Data Lake Gen 2 ke kluster Anda setelah pembuatan blob atau penggantian nama blob.

Pilih tab yang relevan berdasarkan jenis SDK penyimpanan yang digunakan untuk mengunggah blob.

Penyimpanan Blob Azure

Sampel kode berikut menggunakan Azure Blob Storage SDK untuk mengunggah file ke Azure Blob Storage. Unggahan ini memicu koneksi data Event Grid, yang menyerap data ke 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();

Catatan

Azure Data Explorer tidak akan menghapus penyerapan postingan blob. Pertahankan blob selama tiga hingga lima hari dengan menggunakan siklus hidup penyimpanan Azure Blob untuk mengelola penghapusan blob.

Azure Data Lake

Sampel kode berikut menggunakan Azure Data Lake SDK untuk mengunggah file ke Data Lake Storage Gen2. Unggahan ini memicu koneksi data Event Grid, yang menyerap data ke 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);

Catatan

• Saat mengunggah file dengan Azure Data Lake SDK, peristiwa pembuatan file awal memiliki ukuran 0, yang diabaikan oleh Azure Data Explorer selama penyerapan data. Untuk memastikan penyerapan yang tepat, atur parameter ke Close true. Parameter ini menyebabkan metode pengunggahan memicu peristiwa FlushAndClose , yang menunjukkan bahwa pembaruan akhir telah dibuat dan aliran file ditutup.
• Untuk mengurangi lalu lintas yang berasal dari Event Grid dan pemrosesan berikutnya saat menyerap peristiwa ke Azure Data Explorer, sebaiknya filter kunci data.api untuk hanya menyertakan peristiwa FlushAndClose , sehingga menghapus peristiwa pembuatan file dengan ukuran 0. Untuk informasi selengkapnya tentang pembersihan, lihat Metode flush Azure Data Lake.

Mengganti nama blob

Di ADLSv2, Anda dapat mengganti nama direktori. Namun, penting untuk dicatat bahwa mengganti nama direktori tidak memicu peristiwa yang diganti namanya blob atau memulai penyerapan blob yang terkandung dalam direktori. Jika Anda ingin memastikan penyerapan blob setelah mengganti nama direktori, Anda harus langsung mengganti nama blob individual dalam direktori.

Sampel kode berikut menunjukkan cara mengganti nama blob di akun penyimpanan ADLSv2.

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

Catatan

Jika Anda menentukan filter untuk melacak subjek tertentu saat membuat koneksi data atau saat membuat sumber daya Event Grid secara manual, filter ini diterapkan pada jalur file tujuan.

Catatan

Memicu penyerapan setelah CopyBlob operasi tidak didukung untuk akun penyimpanan yang mengaktifkan fitur namespace hierarkis pada akun tersebut.

Menghapus koneksi data Event Grid

C#

Untuk menghapus koneksi Event Grid, jalankan perintah berikut:

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

Python

Untuk menghapus koneksi Event Grid, jalankan perintah berikut:

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)

Konten terkait

• Memproses data dari pusat aktivitas Anda menggunakan Azure Stream Analytics