Vytvoření datového připojení služby Event Hubs pro Azure Data Explorer

Článek
05/03/2023

Azure Data Explorer nabízí příjem dat ze služby Event Hubs, platformu pro streamování velkých objemů dat a službu pro příjem událostí. Event Hubs může zpracovávat miliony událostí za sekundu téměř v reálném čase.

V tomto článku se připojíte k centru událostí a ingestujete data do Azure Data Exploreru. Přehled ingestování ze služby Event Hubs najdete v tématu Datové připojení služby Azure Event Hubs.

Požadavky

Předplatné Azure. Vytvořte bezplatný účet Azure.
Cluster a databáze Azure Data Exploreru. Vytvořte cluster a databázi.
Cílová tabulka. Vytvořte tabulku nebo použijte existující tabulku.
Mapování příjmu dat pro tabulku.
Centrum událostí s daty pro příjem dat.

Vytvoření datového připojení centra událostí

V této části vytvoříte propojení mezi centrem událostí a tabulkou Azure Data Exploreru. Pokud je toto připojení na místě, data se přenášejí z centra událostí do cílové tabulky. Pokud se centrum událostí přesune do jiného prostředku nebo předplatného, budete muset připojení aktualizovat nebo znovu vytvořit.

Na webu Azure Portal přejděte do clusteru a vyberte Databáze. Pak vyberte databázi, která obsahuje cílovou tabulku.
V nabídce vlevo vyberte Příjem dat. Potom v horním panelu vyberte Přidat datové připojení.

Vyplňte formulář následujícími informacemi a pak vyberte Vytvořit.

Nastavení	Navrhovaná hodnota	Popis pole
Název datového připojení	test-hub-connection	Název připojení, které chcete vytvořit v Azure Data Exploreru
Předplatné		ID předplatného, ve kterém se nachází prostředek centra událostí.
Obor názvů centra událostí	Jedinečný název oboru názvů	Název, který jste zvolili dříve a který identifikuje váš obor názvů
Centrum událostí	test-hub	Centrum událostí, které jste vytvořili
Skupina uživatelů	test-group	Skupina uživatelů, kterou jste definovali v centrum událostí, které jste vytvořili
Vlastnosti systému událostí	Výběr relevantních vlastností	Systémové vlastnosti centra událostí. Pokud existuje více záznamů na zprávu události, vlastnosti systému se přidají do prvního záznamu. Při přidávání systémových vlastností vytvořte nebo aktualizujte schéma tabulky a mapování tak, aby zahrnovalo vybrané vlastnosti.
Komprese	Nic	Typ komprese datové části centra událostí. Podporované typy komprese: None, Gzip.
Spravovaná identita (doporučeno)	Přiřazeno systémem	Spravovaná identita používaná clusterem Průzkumníka dat pro přístup ke čtení z centra událostí. Ke kontrole přístupu k centru událostí doporučujeme používat spravované identity. Poznámka: Po vytvoření datového připojení: * Identity přiřazené systémem se automaticky vytvoří, pokud neexistují. * Spravovaná identita se automaticky přiřadí roli příjemce dat služby Azure Event Hubs a přidá se do clusteru Průzkumníka dat. Doporučujeme ověřit, jestli byla role přiřazená a že se identita přidala do clusteru.

Poznámka:

Pokud máte existující datové připojení, které nepoužívá spravované identity, doporučujeme ho aktualizovat na používání spravovaných identit.

Následující kroky vás provedou vytvořením připojení centra událostí prostřednictvím průvodce příjmem dat ve webovém uživatelském rozhraní Azure Data Exploreru.

Poznámka:

Chcete-li použít průvodce, musíte mít alespoň oprávnění uživatele databáze.
Pokud chcete povolit přístup mezi clusterem a účtem úložiště bez veřejného přístupu, přečtěte si téma Vytvoření spravovaného privátního koncového bodu.
Cluster a centrum událostí by měly být přidružené ke stejným tenantům. Pokud ne, použijte některou z možností sady SDK, jako je C# nebo Python.

Na kartě Data ve webovém uživatelském rozhraní Azure Data Exploreru vyberte z karty Centra událostí příjem dat Ingest.
Otevře se okno Ingestování dat s vybranou kartou Cíl . Pole Cluster a Databáze jsou automaticky vyplněna. V rozevíracích nabídkách můžete vybrat jiný cluster nebo databázi.
V části Tabulka vyberte Nová tabulka a zadejte název nové tabulky. Případně můžete použít existující tabulku.
Vyberte Další: Zdroj.
V části Typ zdroje vyberte Centrum událostí.

V části Datové připojení vyplňte následující pole a vyberte Další: Schéma.

Nastavení	Navrhovaná hodnota	Popis pole
Předplatné		ID předplatného, ve kterém se nachází prostředek centra událostí.
Obor názvů centra událostí		Název, který identifikuje váš obor názvů.
Centrum událostí		Centrum událostí, které chcete použít.
Název datového připojení	TestDataConnection	Název, který identifikuje vaše datové připojení.
Skupina uživatelů		Skupina příjemců definovaná ve vašem centru událostí.
Komprese		Typ komprese datové části centra událostí.
Vlastnosti systému událostí	Výběr relevantních vlastností	Systémové vlastnosti centra událostí. Pokud existuje více záznamů na zprávu události, vlastnosti systému se přidají do prvního záznamu. Při přidávání systémových vlastností vytvořte nebo aktualizujte schéma tabulky a mapování tak, aby zahrnovalo vybrané vlastnosti.
Počáteční datum načítání událostí	Koordinovaný univerzální čas (UTC)	Datové připojení načte existující události služby Event Hubs vytvořené po počátečním datu načtení události. Je možné načíst pouze události, které uchovává doba uchovávání služby Event Hubs. Pokud není zadané počáteční datum načtení události, je výchozím časem čas vytvoření datového připojení.

Nastavte zásady příjmu dat. Pokud je pro cluster povolené streamování , můžete vybrat příjem dat streamování. Pokud pro cluster není povolené streamování, nastavte čas dávkování. U služby Event Hubs je doporučená doba dávkování 30 sekund.
Vyberte formát dat. U dat ve formátu CSV ignorujte první záznam a ignorujte řádek záhlaví souboru. U dat ve formátu JSON vyberte Ignorovat chyby formátu dat a ingestujte data ve formátu JSON nebo ponechte nevybraná a ingestujte data ve formátu multijson. Výběrem vnořených úrovní určete rozdělení dat sloupce tabulky.
Pokud data, která vidíte v okně náhledu, nejsou dokončená, možná budete potřebovat další data k vytvoření tabulky se všemi potřebnými datovými poli. K načtení nových dat z centra událostí použijte následující příkazy:
- Zahodit a načíst nová data: zahodí zobrazená data a vyhledá nové události.
- Načtení dalších dat: Vyhledá další události kromě již nalezených událostí.
Poznámka:

Pokud chcete zobrazit náhled dat, musí vaše centrum událostí odesílat události.
Vyberte Další: Souhrn.
V vytvořeném okně centra událostí jsou všechny kroky označené zelenými značkami zaškrtnutí, jakmile se úspěšně dokončí vytváření.

Nainstalujte balíček NuGet Microsoft.Azure.Management.Kusto.
Vytvořte objekt zabezpečení aplikace Azure AD, který se použije k ověřování. Budete potřebovat ID adresáře (tenanta), ID aplikace a tajný klíč klienta.

Spusťte následující kód.

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

Nastavení	Navrhovaná hodnota	Popis pole
tenantId	xxxxxxxx-xxxxx-xxxx-xxxx-xxxxxxxxx	ID vašeho tenanta Označuje se také jako ID adresáře.
subscriptionId	xxxxxxxx-xxxxx-xxxx-xxxx-xxxxxxxxx	ID předplatného, které používáte k vytvoření prostředku.
clientId	xxxxxxxx-xxxxx-xxxx-xxxx-xxxxxxxxx	ID klienta aplikace, která má přístup k prostředkům ve vašem tenantovi.
clientSecret	PlaceholderClientSecret	Tajný klíč klienta aplikace, která má přístup k prostředkům ve vašem tenantovi.
resourceGroupName	testrg	Název skupiny prostředků obsahující váš cluster.
clusterName	mykustocluster	Název clusteru.
databaseName	mykustodatabase	Název cílové databáze v clusteru
eventHubConnectionName	myeventhubconnect	Požadovaný název datového připojení.
tableName	StormEvents	Název cílové tabulky v cílové databázi.
mappingRuleName	StormEvents_CSV_Mapping	Název mapování sloupců související s cílovou tabulkou.
dataFormat	csv	Formát dat zprávy.
eventHubResourceId	ID prostředku	ID prostředku vašeho centra událostí, které obsahuje data pro příjem dat.
consumerGroup	$Default	Skupina příjemců vašeho centra událostí.
location	USA – střed	Umístění prostředku datového připojení
komprese	Gzip nebo None	Typ komprese dat.
databaseRouting	Více nebo jednoúčelový	Směrování databáze pro připojení. Pokud nastavíte hodnotu Single, datové připojení se směruje do izolované databáze v clusteru, jak je uvedeno v nastavení databaseName . Pokud nastavíte hodnotu na Více, můžete výchozí cílovou databázi přepsat pomocí vlastnosti Příjem dat databáze. Další informace naleznete v tématu Směrování událostí.

Nainstalujte požadované knihovny.

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

Vytvořte objekt zabezpečení aplikace Azure AD, který se použije k ověřování. Budete potřebovat ID adresáře (tenanta), ID aplikace a tajný klíč klienta.

Spusťte následující kód.

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

Nastavení	Navrhovaná hodnota	Popis pole
tenant_id	xxxxxxxx-xxxxx-xxxx-xxxx-xxxxxxxxx	ID vašeho tenanta Označuje se také jako ID adresáře.
subscriptionId	xxxxxxxx-xxxxx-xxxx-xxxx-xxxxxxxxx	ID předplatného, které používáte k vytvoření prostředku.
client_id	xxxxxxxx-xxxxx-xxxx-xxxx-xxxxxxxxx	ID klienta aplikace, která má přístup k prostředkům ve vašem tenantovi.
tajný klíč klienta	xxxxxxxxxxxxxxxx	Tajný klíč klienta aplikace, která má přístup k prostředkům ve vašem tenantovi.
resource_group_name	myresourcegroup	Název skupiny prostředků obsahující váš cluster.
cluster_name	mycluster	Název clusteru.
database_name	mydatabase	Název cílové databáze v clusteru
data_connection_name	myeventhubconnect	Požadovaný název datového připojení.
table_name	moje tabulka	Název cílové tabulky v cílové databázi.
mapping_rule_name	mytablemappingrule	Název mapování sloupců související s cílovou tabulkou.
data_format	csv	Formát dat zprávy.
event_hub_resource_id	ID prostředku	ID prostředku vašeho centra událostí, které obsahuje data pro příjem dat.
consumer_group	$Default	Skupina příjemců vašeho centra událostí.
location	USA – střed	Umístění prostředku datového připojení
databaseRouting	Více nebo jednoúčelový	Směrování databáze pro připojení. Pokud nastavíte hodnotu Single, datové připojení se směruje do izolované databáze v clusteru, jak je uvedeno v nastavení databaseName . Pokud nastavíte hodnotu na Více, můžete výchozí cílovou databázi přepsat pomocí vlastnosti Příjem dat databáze. Další informace naleznete v tématu Směrování událostí.

Následující příklad ukazuje šablonu Azure Resource Manageru pro přidání datového připojení služby Event Hubs. Šablonu můžete upravit a nasadit na webu Azure Portal pomocí formuláře.

{
    "$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')]"
            }
        }
    ]
}

Odebrání datového připojení centra událostí

Pokud chcete odebrat připojení centra událostí z webu Azure Portal, postupujte takto:

Přejděte do clusteru. V nabídce vlevo vyberte Databáze. Pak vyberte databázi, která obsahuje cílovou tabulku.
V nabídce vlevo vyberte Datová připojení. Potom zaškrtněte políčko vedle příslušného datového připojení centra událostí.
V horním řádku nabídek vyberte Odstranit.

Pokud chcete odebrat připojení centra událostí, spusťte následující příkaz:

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

Pokud chcete odebrat připojení centra událostí, spusťte následující příkaz:

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)

Další kroky

Dotazování na data s využitím Azure Data Exploreru

Sdílet prostřednictvím

Vytvoření datového připojení služby Event Hubs pro Azure Data Explorer

Požadavky

Vytvoření datového připojení centra událostí

Odebrání datového připojení centra událostí

Další kroky

Další materiály