Create an Event Hubs data connection for Azure Data Explorer

Article
05/03/2023

Azure Data Explorer offers ingestion from Event Hubs, a big data streaming platform and event ingestion service. Event Hubs can process millions of events per second in near real time.

In this article, you'll connect to an event hub and ingest data into Azure Data Explorer. For an overview on ingesting from Event Hubs, see Azure Event Hubs data connection.

Prerequisites

An Azure subscription. Create a free Azure account.
An Azure Data Explorer cluster and database. Create a cluster and database.
A destination table. Create a table or use an existing table.
An ingestion mapping for the table.
An event hub with data for ingestion.

Create an event hub data connection

In this section, you'll establish a connection between the event hub and your Azure Data Explorer table. As long as this connection is in place, data is transmitted from the event hub into your target table. If the event hub is moved to a different resource or subscription, you'll need to update or recreate the connection.

In the Azure portal, go to your cluster and select Databases. Then, select the database that contains your target table.
From the left menu, select Data ingestion. Then, in the top bar, select Add data connection.

Fill out the form with the following information, and then select Create.

Setting	Suggested value	Field description
Data connection name	test-hub-connection	The name of the connection you want to create in Azure Data Explorer.
Subscription		The subscription ID where the event hub resource is located.
Event hub namespace	A unique namespace name	The name you chose earlier that identifies your namespace.
Event hub	test-hub	The event hub you created.
Consumer group	test-group	The consumer group defined in the event hub you created.
Event system properties	Select relevant properties	The event hub system properties. If there are multiple records per event message, the system properties are added to the first record. When adding system properties, create or update table schema and mapping to include the selected properties.
Compression	None	The compression type of the event hub messages payload. Supported compression types: None, Gzip.
Managed Identity (recommended)	System-assigned	The managed identity used by the Data Explorer cluster for access to read from the event hub. We recommend using managed identities to control access to your event hub. Note: When the data connection is created: * System-assigned identities are automatically created if they don't exist * The managed identity is automatically assigned the Azure Event Hubs Data Receiver role and is added to your Data Explorer cluster. We recommend verifying that the role was assigned and that the identity was added to the cluster.

Note

If you have an existing data connection that is not using managed identities, we recommend updating it to use managed identities.

The following steps will guide you through creating an event hub connection through the ingestion wizard in the Azure Data Explorer web UI.

Note

To use the wizard, you must have at least Database User permissions.
To enable access between a cluster and a storage account without public access, see Create a Managed Private Endpoint.
The cluster and event hub should be associated with the same tenants. If not, use one of the SDK options, such as C# or Python.

From the Data tab of the Azure Data Explorer web UI, select Ingest from the Ingest data from Event Hub card.
The Ingest data window opens with the Destination tab selected. The Cluster and Database fields are autopopulated. You may select a different cluster or database from the drop-down menus.
Under Table, select New table and enter a name for the new table. Alternatively, use an existing table.
Select Next: Source.
Under Source type, select Event Hub.

Under Data Connection, fill in the following fields and select Next: Schema.

Setting	Suggested value	Field description
Subscription		The subscription ID where the event hub resource is located.
Event hub namespace		The name that identifies your namespace.
Event hub		The event hub you wish to use.
Data connection name	TestDataConnection	The name that identifies your data connection.
Consumer group		The consumer group defined in your event hub.
Compression		The compression type of the event hub messages payload.
Event system properties	Select relevant properties	The event hub system properties. If there are multiple records per event message, the system properties are added to the first one. When adding system properties, create or update table schema and mapping to include the selected properties.
Event retrieval start date	Coordinated Universal Time (UTC)	The data connection retrieves existing Event Hubs events created after the Event retrieval start date. Only events retained by Event Hubs's retention period can be retrieved. If the Event retrieval start date isn't specified, the default time is the time at which the data connection is created.

Set the ingestion policy. If streaming is enabled for the cluster, you can select Streaming ingestion. If streaming isn't enabled for the cluster, set the Batching time. For Event Hubs, the recommended batching time is 30 seconds.
Select the Data format. For CSV-formatted data, Ignore the first record to ignore the heading row of the file. For JSON-formatted data, select Ignore data format errors to ingest the data in JSON format or leave unselected to ingest the data in multijson format. Select the Nested levels to determine the table column data division.
If the data you see in the preview window isn't complete, you may need more data to create a table with all necessary data fields. Use the following commands to fetch new data from your event hub:
- Discard and fetch new data: discards the data presented and searches for new events.
- Fetch more data: Searches for more events in addition to the events already found.
Note

To see a preview of your data, your event hub must be sending events.
Select Next: Summary.
In the Continuous ingestion from Event Hub established window, all steps are marked with green check marks when establishment finishes successfully.

Install the Microsoft.Azure.Management.Kusto NuGet package.
Create an Azure AD application principal to use for authentication. You'll need the directory (tenant) ID, application ID, and client secret.

Run the following code.

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

Setting	Suggested value	Field description
tenantId	xxxxxxxx-xxxxx-xxxx-xxxx-xxxxxxxxx	Your tenant ID. Also known as directory ID.
subscriptionId	xxxxxxxx-xxxxx-xxxx-xxxx-xxxxxxxxx	The subscription ID that you use for resource creation.
clientId	xxxxxxxx-xxxxx-xxxx-xxxx-xxxxxxxxx	The client ID of the application that can access resources in your tenant.
clientSecret	PlaceholderClientSecret	The client secret of the application that can access resources in your tenant.
resourceGroupName	testrg	The name of the resource group containing your cluster.
clusterName	mykustocluster	The name of your cluster.
databaseName	mykustodatabase	The name of the target database in your cluster.
eventHubConnectionName	myeventhubconnect	The desired name of your data connection.
tableName	StormEvents	The name of the target table in the target database.
mappingRuleName	StormEvents_CSV_Mapping	The name of your column mapping related to the target table.
dataFormat	csv	The data format of the message.
eventHubResourceId	Resource ID	The resource ID of your event hub that holds the data for ingestion.
consumerGroup	$Default	The consumer group of your event hub.
location	Central US	The location of the data connection resource.
compression	Gzip or None	The type of data compression.
databaseRouting	Multi or Single	The database routing for the connection. If you set the value to Single, the data connection is 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 ingestion property. For more information, see Events routing.

Install the required libraries.

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

Create an Azure AD application principal to use for authentication. You'll need the directory (tenant) ID, application ID, and client secret.

Run the following code.

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

Setting	Suggested value	Field description
tenant_id	xxxxxxxx-xxxxx-xxxx-xxxx-xxxxxxxxx	Your tenant ID. Also known as directory ID.
subscriptionId	xxxxxxxx-xxxxx-xxxx-xxxx-xxxxxxxxx	The subscription ID that you use for resource creation.
client_id	xxxxxxxx-xxxxx-xxxx-xxxx-xxxxxxxxx	The client ID of the application that can access resources in your tenant.
client_secret	xxxxxxxxxxxxxx	The client secret of the application that can access resources in your tenant.
resource_group_name	myresourcegroup	The name of the resource group containing your cluster.
cluster_name	mycluster	The name of your cluster.
database_name	mydatabase	The name of the target database in your cluster.
data_connection_name	myeventhubconnect	The desired name of your data connection.
table_name	mytable	The name of the target table in the target database.
mapping_rule_name	mytablemappingrule	The name of your column mapping related to the target table.
data_format	csv	The data format of the message.
event_hub_resource_id	Resource ID	The resource ID of your event hub that holds the data for ingestion.
consumer_group	$Default	The consumer group of your event hub.
location	Central US	The location of the data connection resource.
databaseRouting	Multi or Single	The database routing for the connection. If you set the value to Single, the data connection is 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 ingestion property. For more information, see Events routing.

The following example shows an Azure Resource Manager template for adding an Event Hubs data connection. You can edit and deploy the template in the Azure portal by using the form.

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

Remove an event hub data connection

To remove the event hub connection from the Azure portal, do the following:

Go to your cluster. From the left menu, select Databases. Then, select the database that contains the target table.
From the left menu, select Data connections. Then, select the checkbox next to the relevant event hub data connection.
From the top menu bar, select Delete.

To remove the event hub connection, run the following command:

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