Tworzenie połączenia danych usługi Event Hubs dla usługi Azure Data Explorer

Artykuł
05/03/2023

Usługa Azure Data Explorer oferuje pozyskiwanie z usługi Event Hubs, platformy przesyłania strumieniowego danych big data i usługi pozyskiwania zdarzeń. Usługa Event Hubs może przetwarzać miliony zdarzeń na sekundę niemal w czasie rzeczywistym.

W tym artykule połączysz się z centrum zdarzeń i pozyskujesz dane z usługą Azure Data Explorer. Aby zapoznać się z omówieniem pozyskiwania z usługi Event Hubs, zobacz Azure Event Hubs połączenie danych.

Wymagania wstępne

Subskrypcja platformy Azure. Utwórz bezpłatne konto platformy Azure.
Baza danych i klaster usługi Azure Data Explorer. Utwórz klaster i bazę danych.
Tabela docelowa. Utwórz tabelę lub użyj istniejącej tabeli.
Mapowanie pozyskiwania dla tabeli.
Centrum zdarzeń z danymi do pozyskiwania.

Tworzenie połączenia danych centrum zdarzeń

W tej sekcji ustanowisz połączenie między centrum zdarzeń a tabelą usługi Azure Data Explorer. O ile to połączenie jest na miejscu, dane są przesyłane z centrum zdarzeń do tabeli docelowej. Jeśli centrum zdarzeń zostanie przeniesione do innego zasobu lub subskrypcji, musisz zaktualizować lub ponownie utworzyć połączenie.

W Azure Portal przejdź do klastra i wybierz pozycję Bazy danych. Następnie wybierz bazę danych zawierającą tabelę docelową.
W menu po lewej stronie wybierz pozycję Pozyskiwanie danych. Następnie na górnym pasku wybierz pozycję Dodaj połączenie danych.

Wypełnij formularz następującymi informacjami, a następnie wybierz pozycję Utwórz.

Ustawienie	Sugerowana wartość	Opis pola
Nazwa połączenia danych	test-hub-connection	Nazwa połączenia, które chcesz utworzyć w usłudze Azure Data Explorer.
Subskrypcja		Identyfikator subskrypcji, w której znajduje się zasób centrum zdarzeń.
Przestrzeń nazw centrum zdarzeń	Unikatowa nazwa przestrzeni nazw	Wybrana wcześniej nazwa, która identyfikuje Twoją przestrzeń nazw.
Centrum zdarzeń	test-hub	Utworzone przez Ciebie centrum zdarzeń.
Grupa konsumentów	test-group	Grupa konsumentów zdefiniowana w utworzonym przez Ciebie centrum zdarzeń.
Właściwości systemu zdarzeń	Wybieranie odpowiednich właściwości	Właściwości systemu centrum zdarzeń. Jeśli istnieje wiele rekordów na komunikat zdarzenia, właściwości systemowe są dodawane do pierwszego rekordu. Podczas dodawania właściwości systemowych utwórz lub zaktualizuj schemat tabeli i mapowanie , aby uwzględnić wybrane właściwości.
Kompresja	Brak	Typ kompresji ładunku komunikatów centrum zdarzeń. Obsługiwane typy kompresji: Brak, Gzip.
Tożsamość zarządzana (zalecana)	Przypisane przez system	Tożsamość zarządzana używana przez klaster Data Explorer na potrzeby dostępu do odczytu z centrum zdarzeń. Zalecamy używanie tożsamości zarządzanych do kontrolowania dostępu do centrum zdarzeń. Uwaga: Po utworzeniu połączenia danych: * Tożsamości przypisane przez system są tworzone automatycznie, jeśli nie istnieją * Tożsamość zarządzana jest automatycznie przypisywana do roli odbiornika danych Azure Event Hubs i jest dodawana do klastra Data Explorer. Zalecamy sprawdzenie, czy rola została przypisana i czy tożsamość została dodana do klastra.

Uwaga

Jeśli masz istniejące połączenie danych, które nie korzysta z tożsamości zarządzanych, zalecamy zaktualizowanie go w celu korzystania z tożsamości zarządzanych.

Poniższe kroki przeprowadzą Cię przez proces tworzenia połączenia centrum zdarzeń za pośrednictwem kreatora pozyskiwania w internetowym interfejsie użytkownika usługi Azure Data Explorer.

Uwaga

Aby użyć kreatora, musisz mieć co najmniej uprawnienia użytkownika bazy danych .
Aby włączyć dostęp między klastrem a kontem magazynu bez dostępu publicznego, zobacz Tworzenie zarządzanego prywatnego punktu końcowego.
Klaster i centrum zdarzeń powinny być skojarzone z tymi samymi dzierżawami. Jeśli nie, użyj jednej z opcji zestawu SDK, takich jak C# lub Python.

Na karcie Daneinternetowego interfejsu użytkownika usługi Azure Data Explorer wybierz pozycję Pozyskiwanie z karty Pozyskiwanie danych z centrum zdarzeń.
Zostanie otwarte okno Pozyskiwanie danych z wybraną kartą Miejsce docelowe . Pola Klaster i Baza danych są wypełniane automatycznie. Możesz wybrać inny klaster lub bazę danych z menu rozwijanych.
W obszarze Tabela wybierz pozycję Nowa tabela i wprowadź nazwę nowej tabeli. Alternatywnie użyj istniejącej tabeli.
Wybierz pozycję Dalej: Źródło.
W obszarze Typ źródła wybierz pozycję Centrum zdarzeń.

W obszarze Połączenie danych wypełnij następujące pola i wybierz pozycję Dalej: Schemat.

Ustawienie	Sugerowana wartość	Opis pola
Subskrypcja		Identyfikator subskrypcji, w której znajduje się zasób centrum zdarzeń.
Przestrzeń nazw centrum zdarzeń		Nazwa identyfikująca przestrzeń nazw.
Centrum zdarzeń		Centrum zdarzeń, którego chcesz użyć.
Nazwa połączenia danych	TestDataConnection	Nazwa identyfikująca połączenie danych.
Grupa konsumentów		Grupa odbiorców zdefiniowana w centrum zdarzeń.
Kompresja		Typ kompresji ładunku komunikatów centrum zdarzeń.
Właściwości systemu zdarzeń	Wybieranie odpowiednich właściwości	Właściwości systemu centrum zdarzeń. Jeśli istnieje wiele rekordów na komunikat zdarzenia, właściwości systemowe są dodawane do pierwszego. Podczas dodawania właściwości systemowych utwórz lub zaktualizuj schemat tabeli i mapowanie , aby uwzględnić wybrane właściwości.
Data rozpoczęcia pobierania zdarzeń	Uniwersalny czas koordynowany (UTC)	Połączenie danych pobiera istniejące zdarzenia usługi Event Hubs utworzone po dacie rozpoczęcia pobierania zdarzeń. Można pobierać tylko zdarzenia przechowywane przez okres przechowywania usługi Event Hubs. Jeśli data rozpoczęcia pobierania zdarzeń nie jest określona, domyślny czas to godzina utworzenia połączenia danych.

Ustaw zasady pozyskiwania. Jeśli przesyłanie strumieniowe jest włączone dla klastra, możesz wybrać pozycję Pozyskiwanie przesyłania strumieniowego. Jeśli przesyłanie strumieniowe nie jest włączone dla klastra, ustaw czas przetwarzania wsadowego. W przypadku usługi Event Hubs zalecany czas dzielenia na partie wynosi 30 sekund.
Wybierz format danych. W przypadku danych sformatowanych w formacie CSV zignoruj pierwszy rekord , aby zignorować wiersz nagłówka pliku. W przypadku danych sformatowanych w formacie JSON wybierz pozycję Ignoruj błędy formatu danych , aby pozyskać dane w formacie JSON lub pozostawić niezaznaczone, aby pozyskać dane w formacie wielossonowym. Wybierz poziomy zagnieżdżone , aby określić podział danych kolumny tabeli.
Jeśli dane widoczne w oknie podglądu nie są kompletne, może być konieczne utworzenie tabeli ze wszystkimi wymaganymi polami danych. Użyj następujących poleceń, aby pobrać nowe dane z centrum zdarzeń:
- Odrzuć i pobierz nowe dane: odrzuca prezentowane dane i wyszukuje nowe zdarzenia.
- Pobieranie większej ilości danych: wyszukuje więcej zdarzeń oprócz znalezionych już zdarzeń.
Uwaga

Aby wyświetlić podgląd danych, centrum zdarzeń musi wysyłać zdarzenia.
Wybierz pozycję Dalej: Podsumowanie.
W oknie ciągłego pozyskiwania z centrum zdarzeń wszystkie kroki są oznaczone zielonymi znacznikami wyboru po pomyślnym zakończeniu ustanowienia.

Zainstaluj pakiet NuGet Microsoft.Azure.Management.Kusto.
Utwórz jednostkę aplikacji Azure AD do użycia na potrzeby uwierzytelniania. Będziesz potrzebować identyfikatora katalogu (dzierżawy), identyfikatora aplikacji i klucza tajnego klienta.

Uruchom następujący kod.

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

Ustawienie	Sugerowana wartość	Opis pola
tenantId	xxxxxxxx-xxxxx-xxxx-xxxx-xxxxxxxxx	Identyfikator dzierżawy. Znany również jako identyfikator katalogu.
subscriptionId	xxxxxxxx-xxxxx-xxxx-xxxx-xxxxxxxxx	Identyfikator subskrypcji używany do tworzenia zasobów.
clientId	xxxxxxxx-xxxxx-xxxx-xxxx-xxxxxxxxx	Identyfikator klienta aplikacji, która może uzyskiwać dostęp do zasobów w dzierżawie.
clientSecret	Symbol zastępczyClientSecret	Wpis tajny klienta aplikacji, która może uzyskiwać dostęp do zasobów w dzierżawie.
resourceGroupName	testrg	Nazwa grupy zasobów zawierającej klaster.
clusterName	mykustocluster	Nazwa klastra.
databaseName	mykustodatabase	Nazwa docelowej bazy danych w klastrze.
eventHubConnectionName	myeventhubconnect	Żądana nazwa połączenia danych.
tableName	StormEvents	Nazwa tabeli docelowej w docelowej bazie danych.
mappingRuleName	StormEvents_CSV_Mapping	Nazwa mapowania kolumn powiązanych z tabelą docelową.
Dataformat	Csv	Format danych komunikatu.
eventHubResourceId	Identyfikator zasobu	Identyfikator zasobu centrum zdarzeń, który przechowuje dane do pozyskiwania.
consumerGroup	$Default	Grupa odbiorców centrum zdarzeń.
location	Środkowe stany USA	Lokalizacja zasobu połączenia danych.
kompresja	Gzip lub Brak	Typ kompresji danych.
databaseRouting	Wiele lub pojedyncza	Routing bazy danych dla połączenia. Jeśli ustawisz wartość Na pojedynczą, połączenie danych jest kierowane do pojedynczej bazy danych w klastrze, jak określono w ustawieniu databaseName . Jeśli ustawisz wartość Na wiele, możesz przesłonić domyślną docelową bazę danych przy użyciu właściwości pozyskiwaniabazy danych. Aby uzyskać więcej informacji, zobacz Routing zdarzeń.

Zainstaluj wymagane biblioteki.

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

Utwórz jednostkę aplikacji Azure AD do użycia na potrzeby uwierzytelniania. Będziesz potrzebować identyfikatora katalogu (dzierżawy), identyfikatora aplikacji i klucza tajnego klienta.

Uruchom następujący kod.

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

Ustawienie	Sugerowana wartość	Opis pola
tenant_id	xxxxxxxx-xxxxx-xxxx-xxxx-xxxxxxxxx	Identyfikator dzierżawy. Znany również jako identyfikator katalogu.
subscriptionId	xxxxxxxx-xxxxx-xxxx-xxxx-xxxxxxxxx	Identyfikator subskrypcji używany do tworzenia zasobów.
client_id	xxxxxxxx-xxxxx-xxxx-xxxx-xxxxxxxxx	Identyfikator klienta aplikacji, która może uzyskiwać dostęp do zasobów w dzierżawie.
client_secret	xxxxxxxxxxxxxx	Klucz tajny klienta aplikacji, który może uzyskiwać dostęp do zasobów w dzierżawie.
resource_group_name	myresourcegroup	Nazwa grupy zasobów zawierającej klaster.
cluster_name	mycluster	Nazwa klastra.
Nazwa_bazy_danych	Mojabazadanych	Nazwa docelowej bazy danych w klastrze.
data_connection_name	myeventhubconnect	Żądana nazwa połączenia danych.
Nazwa_tabeli	Mytable	Nazwa tabeli docelowej w docelowej bazie danych.
mapping_rule_name	mytablemappingrule	Nazwa mapowania kolumny powiązana z tabelą docelową.
data_format	Csv	Format danych komunikatu.
event_hub_resource_id	Identyfikator zasobu	Identyfikator zasobu centrum zdarzeń, który przechowuje dane pozyskiwania.
consumer_group	$Default	Grupa odbiorców centrum zdarzeń.
location	Środkowe stany USA	Lokalizacja zasobu połączenia danych.
databaseRouting	Wiele lub pojedynczy	Routing bazy danych dla połączenia. Jeśli ustawisz wartość Pojedyncza, połączenie danych zostanie przekierowane do pojedynczej bazy danych w klastrze, jak określono w ustawieniu databaseName . Jeśli ustawisz wartość Multi, możesz zastąpić domyślną docelową bazę danych przy użyciu właściwości Pozyskiwaniebazy danych. Aby uzyskać więcej informacji, zobacz Routing zdarzeń.

W poniższym przykładzie przedstawiono szablon usługi Azure Resource Manager do dodawania połączenia danych usługi Event Hubs. Szablon można edytować i wdrożyć w Azure Portal przy użyciu formularza.

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

Usuwanie połączenia danych centrum zdarzeń

Aby usunąć połączenie centrum zdarzeń z Azure Portal, wykonaj następujące czynności:

Przejdź do klastra. W menu po lewej stronie wybierz pozycję Bazy danych. Następnie wybierz bazę danych zawierającą tabelę docelową.
W menu po lewej stronie wybierz pozycję Połączenia danych. Następnie zaznacz pole wyboru obok odpowiedniego połączenia danych centrum zdarzeń.
Na górnym pasku menu wybierz pozycję Usuń.

Aby usunąć połączenie centrum zdarzeń, uruchom następujące polecenie:

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