Event Grid-adatkapcsolat létrehozása az Azure Data Explorerhez SDK-kkal

Ebből a cikkből megtudhatja, hogyan lehet blobokat beszúrni a tárfiókból az Azure Data Explorerbe event Grid-adatkapcsolat használatával. Létre fog hozni egy Event Grid-adatkapcsolatot, amely beállít egy Azure Event Grid-előfizetést . Az Event Grid-előfizetés egy Azure Event Hubson keresztül irányítja át az eseményeket a tárfiókból az Azure Data Explorerbe.

Ha tudni szeretné, hogyan hozhatja létre a kapcsolatot az Azure Portalon vagy ARM-sablonnal, olvassa el az Event Grid-adatkapcsolat létrehozása című témakört.

Az Azure Data Explorerbe az Event Gridből való betöltéssel kapcsolatos általános információkért lásd: Csatlakozás az Event Gridhez.

Megjegyzés:

Az Event Grid-kapcsolaton keresztüli legjobb teljesítmény eléréséhez állítsa be a rawSizeBytes betöltési tulajdonságot a blob metaadatain keresztül. További információkért lásd a bevitel tulajdonságait.

A korábbi SDK-verziókon alapuló kódmintákért lásd az archivált cikket.

Előfeltételek

• Egy Azure-előfizetés. Hozzon létre egy ingyenes Azure-fiókot.
• Egy Azure Data Explorer-fürt és adatbázis. Hozzon létre egy klasztert és egy adatbázist.
• Célállomás táblázat. Hozzon létre egy táblát , vagy használjon egy meglévő táblát.
• A táblázat betöltési leképezése .
• Egy tárfiók. Event Grid-értesítési előfizetés beállítható az Azure Storage-fiókokra a BlobStorage, StorageV2, vagy a 2. generációs Data Lake Storage () esetében.
• Regisztrálva van az Event Grid erőforrás-szolgáltató.

Event Grid-adatkapcsolat létrehozása

Ebben a szakaszban kapcsolatot fog létesíteni az Event Grid és az Azure Data Explorer-tábla között.

C#

1. Telepítse a Microsoft.Azure.Management.Kusto NuGet csomagot.

2. Hozzon létre egy Microsoft Entra-alkalmazásnevet a hitelesítéshez. Szüksége lesz a címtár (bérlő) azonosítójára, az alkalmazásazonosítóra és az ügyfél titkos kódjára.

3. Futtassa a következő kódot.

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

   Setting	Ajánlott érték	Mező leírása
   tenantId	xxxxxxxx-xxxxx-xxxx-xxxx-xxxx-xxxxxxxxx	Az Ön bérlőazonosítója. Más néven könyvtárazonosító.
   subscriptionId	xxxxxxxx-xxxxx-xxxx-xxxx-xxxx-xxxxxxxxx	Az erőforrás-létrehozáshoz használt előfizetés-azonosító.
   clientId	xxxxxxxx-xxxxx-xxxx-xxxx-xxxx-xxxxxxxxx	Annak az alkalmazásnak az ügyfélazonosítója, amely hozzáfér a bérlő erőforrásaihoz.
   clientSecret	PlaceholderClientSecret	Az alkalmazás ügyfélkulcsa, amely hozzáférést biztosít a bérlői környezet erőforrásaihoz.
   resourceGroupName	testrg	A fürtöt tartalmazó erőforráscsoport neve.
   klaszternév	mykustocluster	A klaszter neve.
   databaseName	mykustodatabase	A fürt céladatbázisának neve.
   eventGridConnectionName	myeventgridconnect	Az adatkapcsolat kívánt neve.
   tableName	StormEvents	A céltábla neve a céladatbázisban.
   leképezési szabály neve	ViharEsemények_CSV_Térkép	A céltáblához kapcsolódó oszlopleképezés neve.
   adatformátum	csv	Az üzenet adatformátuma.
   eventHubResourceId	Erőforrás-azonosító	Az eseményközpont erőforrás-azonosítója, ahol az Event Grid események küldésére van konfigurálva.
   storageAccountResourceId	Erőforrás-azonosító	A betöltéshez szükséges adatokat tartalmazó tárfiók erőforrás-azonosítója.
   fogyasztói csoport	$Alapértelmezett	Az eseményközpont fogyasztói csoportja.
   hely	USA középső régiója	Az adatkapcsolati erőforrás helye.
   blob tárolási eseménytípus	Microsoft.Storage.BlobCreated	Az esemény típusa, amely bemenetet vált ki. A támogatott események a következők: Microsoft.Storage.BlobCreated vagy Microsoft.Storage.BlobRenamed. A blob átnevezése csak az ADLSv2 storage esetében támogatott.
   adatbázis-irányítás	Több - vagy egyszemélyes	A kapcsolat adatbázis-útválasztása. Ha az értéket Single értékre állítja, az adatkapcsolatot a fürtön belül a databaseName beállításban megadott adatbázishoz irányítják. Ha multi értékre állítja az értéket, felülbírálhatja az alapértelmezett céladatbázist az Adatbázis-betöltési tulajdonság használatával. További információ: Események útválasztása.

Python

1. Telepítse a szükséges kódtárakat.

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

2. Hozzon létre egy Microsoft Entra-alkalmazásnevet a hitelesítéshez. Szüksége lesz a címtár (bérlő) azonosítójára, az alkalmazásazonosítóra és az ügyfél titkos kódjára.

3. Futtassa a következő kódot.

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

   Setting	Ajánlott érték	Mező leírása
   bérlő_azonosító	xxxxxxxx-xxxxx-xxxx-xxxx-xxxx-xxxxxxxxx	Az Ön bérlőazonosítója. Más néven könyvtárazonosító.
   subscription_id	xxxxxxxx-xxxxx-xxxx-xxxx-xxxx-xxxxxxxxx	Az erőforrás-létrehozáshoz használt előfizetés-azonosító.
   client_id	xxxxxxxx-xxxxx-xxxx-xxxx-xxxx-xxxxxxxxx	Annak az alkalmazásnak az ügyfélazonosítója, amely hozzáfér a bérlő erőforrásaihoz.
   client_secret	xxxxxxxxxxxxxx	Az alkalmazás ügyfélkulcsa, amely hozzáférést biztosít a bérlői környezet erőforrásaihoz.
   erőforrás_csoport_neve	testrg	A fürtöt tartalmazó erőforráscsoport neve.
   klaszter_név	mykustocluster	A klaszter neve.
   Adatbázis_név	mykustodatabase	A fürt céladatbázisának neve.
   adatkapcsolat_neve	myeventhubconnect	Az adatkapcsolat kívánt neve.
   table_name	StormEvents	A céltábla neve a céladatbázisban.
   leképezési_szabály_neve	ViharEsemények_CSV_Térkép	A céltáblához kapcsolódó oszlopleképezés neve.
   database_routing	Több - vagy egyszemélyes	A kapcsolat adatbázis-útválasztása. Ha az értéket egyetlen értékre állítja, a rendszer az adatkapcsolatot a fürt egy adott adatbázisához irányítja a databaseName beállításban meghatározott módon. Ha multi értékre állítja az értéket, felülbírálhatja az alapértelmezett céladatbázist az Adatbázis-betöltési tulajdonság használatával. További információ: Események útválasztása.
   adatformátum	csv	Az üzenet adatformátuma.
   event_hub_erőforrás_azonosító	Erőforrás-azonosító	Az eseményközpont erőforrás-azonosítója, ahol az Event Grid események küldésére van konfigurálva.
   storage_account_resource_id	Erőforrás-azonosító	A betöltéshez szükséges adatokat tartalmazó tárfiók erőforrás-azonosítója.
   fogyasztói csoport	$Alapértelmezett	Az eseményközpont fogyasztói csoportja.
   hely	USA középső régiója	Az adatkapcsolati erőforrás helye.
   blob_tároló_esemény_típus	Microsoft.Storage.BlobCreated	Az esemény típusa, amely bemenetet vált ki. A támogatott események a következők: Microsoft.Storage.BlobCreated vagy Microsoft.Storage.BlobRenamed. A blob átnevezése csak az ADLSv2 storage esetében támogatott.

Az Event Grid-adatkapcsolat használata

Ez a szakasz bemutatja, hogyan lehet indítani az adatbetöltést az Azure Blob Storage-ból vagy az Azure Data Lake Gen 2-ből a fürtbe a blob létrehozását vagy átnevezését követően.

Válassza ki a megfelelő lapot a blobok feltöltéséhez használt tárolási SDK típusától függően.

Azure Blob-tároló

Az alábbi kódminta az Azure Blob Storage SDK használatával tölt fel egy fájlt az Azure Blob Storage-ba. A feltöltés aktiválja az Event Grid adatkapcsolatot, amely betölti az adatokat az Azure Data Explorerbe.

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

Megjegyzés:

Az Azure Data Explorer nem törli a blobokat a betöltés után. Tartsa meg a blobokat 3–5 napig az Azure Blob Storage életciklusának használatával a blobok törlésének kezeléséhez.

Azure Data Lake

Az alábbi kódminta az Azure Data Lake SDK használatával tölt fel egy fájlt a Data Lake Storage Gen2-be. A feltöltés aktiválja az Event Grid adatkapcsolatot, amely betölti az adatokat az Azure Data Explorerbe.

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

Megjegyzés:

• Amikor feltölt egy fájlt az Azure Data Lake SDK-val, a kezdeti fájllétrehozási esemény mérete 0, amelyet az Azure Data Explorer figyelmen kívül hagy az adatbetöltés során. A megfelelő betöltés biztosításához állítsa a Close paramétert true értékre. Ez a paraméter egy FlushAndClose eseményt indít el a feltöltési metódusban, ami azt jelzi, hogy a végleges frissítés megtörtént, és a fájlstream bezárult.
• Az Event Gridből érkező forgalom csökkentése és az események Azure Data Explorerbe való betöltésekor történő későbbi feldolgozás érdekében javasoljuk, hogy szűrje a data.api-kulcsot , hogy csak a FlushAndClose-eseményeket tartalmazza, ezzel eltávolítva a 0-s méretű fájllétrehozási eseményeket. A kiürítéssel kapcsolatos további információkért lásd az Azure Data Lake flush metódusát.

Blobok átnevezése

Az ADLSv2-ben átnevezheti a könyvtárakat. Fontos azonban megjegyezni, hogy a címtár átnevezése nem váltja ki a blob átnevezett eseményeit, és nem kezdeményezi a címtárban található blobok betöltését. Ha azt szeretné, hogy a blobok betöltése biztosítva legyen, miután átnevezte a könyvtárat, akkor közvetlenül nevezze át a könyvtárban lévő egyes blobokat.

Az alábbi kódminta bemutatja, hogyan nevezhet át blobokat egy ADLSv2-tárfiókban.

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

Megjegyzés:

Ha meghatározott szűrőket adott témák nyomon követésére az adatkapcsolat létrehozásakor vagy az Event Grid-erőforrások manuális létrehozásakor, ezek a szűrők a célfájl elérési útján lesznek alkalmazva.

Megjegyzés:

A CopyBlob művelet utáni bevitel aktiválása nem lehetséges azoknál a tárfiókoknál, amelyeknél engedélyezve van a hierarchikus névtér funkció.

Fontos

Nagy mértékben elriasztjuk a Storage-események egyéni kódból való létrehozását és az Event Hubsba való küldését. Ha ezt választja, győződjön meg arról, hogy a létrehozott események szigorúan megfelelnek a megfelelő Storage Events-sémának és A JSON formátum specifikációinak.

Event Grid-adatkapcsolat eltávolítása

C#

Az Event Grid-kapcsolat eltávolításához futtassa a következő parancsot:

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

Python

Az Event Grid-kapcsolat eltávolításához futtassa a következő parancsot:

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)

Kapcsolódó tartalom

• Adatok feldolgozása az eseményközpontból az Azure Stream Analytics használatával