Создание подключения к данным сетки событий для Azure Data Explorer

Статья
05/03/2023

В этой статье вы узнаете, как принимать большие двоичные объекты из вашей учетной записи хранения в Azure Data Explorer с помощью подключения к данным Сетки событий. Вы создадите подключение к данным Сетки событий, которое задает подписку Сетка событий Azure. Подписка на сетку событий направляет события из вашей учетной записи хранения в Azure Data Explorer через Центры событий Azure.

Общие сведения о передаче данных в Azure Data Explorer из сетки событий см. в разделе Подключение к сетке событий.

Примечание.

Чтобы обеспечить лучшую производительность при подключении сетки событий, задайте rawSizeBytes свойство приема с помощью метаданных большого двоичного объекта. Дополнительные сведения см. в свойствах приема.

Необходимые компоненты

Подписка Azure. Создайте бесплатную учетную запись Azure.
Кластер и база данных Azure Data Explorer. Создайте кластер и базу данных.
Целевая таблица. Создайте таблицу или используйте существующую таблицу.
Сопоставление приема для таблицы.
Учетная запись хранения. Подписка на уведомление сетки событий может быть задана в учетных записях служба хранилища Azure для BlobStorageData Lake Storage 2-го поколения StorageV2или Data Lake Storage 2-го поколения.

Создание подключения к данным в Сетке событий

В этом разделе описано, как установить соединение между сеткой событий и таблицей Azure Data Explorer.

Под созданным кластером выберите Базы данных>TestDatabase.
Выберите Прием данных>Добавить подключение к данным.

На вкладке Основное выберите тип подключения Хранилище BLOB-объектов и внесите остальные сведения:

Снимок экрана: вкладка

Параметр	Рекомендуемое значение	Описание поля
Имя подключения к данным	test-grid-connection	Имя создаваемого подключения в Azure Data Explorer.
Подписка учетной записи хранения	Ваш идентификатор подписки	Идентификатор подписки, в которой находится ваша учетная запись хранения.
Storage account	gridteststorage1	Имя созданной ранее учетной записи хранения.
Тип события	Созданный BLOB-объект или переименованный BLOB-объект	Тип события, запускающего загрузку. Переименование BLOB-объектов поддерживается только для хранилища ADLSv2. Чтобы переименовать большой двоичный объект, перейдите к большому двоичному объекту в портал Azure, щелкните его правой кнопкой мыши и выберите "Переименовать". Поддерживаемые типы: Microsoft.Storage.BlobCreated или Microsoft.Storage.BlobRenamed.
Создание ресурсов	Автоматически	Определите, хотите ли вы, чтобы Azure Data Explorer создавал для вас подписку на Сетку событий, пространство имен Центров событий и сами Центры событий. Чтобы создать ресурсы вручную, см. раздел Создание ресурсов для приема Сетки событий вручную

Выберите Настройки фильтра, если вы хотите отслеживать определенные предметы. Настройте фильтры для получения уведомлений следующим образом:
- Поле Префикс — это буквальный префикс темы. Поскольку применяемый шаблон — startswith, он может охватывать несколько контейнеров, папок или BLOB-объектов. Подстановочные знаки не допускаются.
  - Чтобы определить фильтр для контейнера больших двоичных объектов, поле должно быть установлено следующим образом: /blobServices/default/containers/[container prefix].
  - Чтобы определить фильтр для префикса большого двоичного объекта (или папки в Azure Data Lake 2 поколения), поле должно быть установлено следующим образом: /blobServices/default/containers/[container name]/blobs/[folder/blob prefix].
- Поле суффикс — это буквальный суффикс большого двоичного объекта. Подстановочные знаки не допускаются.
- Поле С учетом регистра указывает, учитываются ли фильтры префикса и суффикса.
- Дополнительные сведения о фильтрации событий см. в разделе События хранилища больших двоичных объектов.
Выберите Далее: вставить свойства.

Заполните форму, указав следующую информацию. Имена таблиц и отображений чувствительны к регистру:

Снимок экрана: вкладка

Свойства приема

Параметр	Рекомендуемое значение	Описание поля
Разрешить маршрутизацию данных в другие базы данных (подключение данных к нескольким базам данных)	Не разрешать	Включите этот параметр, если требуется переопределить целевую базу данных по умолчанию, связанную с подключением к данным. Дополнительные сведения о маршрутизации в базы данных см. в разделе Маршрутизация событий.
Имя таблицы	TestTable	Таблица, созданная в базе данных TestDatabase.
Формат данных	JSON	Поддерживаемые форматы: Avro, CSV, JSON, MULTILINE JSON, ORC, PARQUET, PSV, SCSV, SOHSV, TSV, TXT, TSVE, APACHEAVRO, RAW и W3CLOG. Поддерживаются форматы сжатия ZIP и GZIP.
Сопоставление	TestMapping	Сопоставление, созданное в TestDatabase, которое сопоставляет входящие данные с именами столбцов и типами данных TestTable. Если значение не указано, используется сопоставление данных удостоверений, полученное из схемы таблицы.
Расширенные настройки	У моих данных есть заголовки	Игнорирует заголовки. Поддерживается для файлов типа *SV.

Примечание.

Вы не обязаны указывать все значения для параметров маршрутизации по умолчанию. Принимаются также частично заполненные параметры.

Выберите Далее: Проверка и создание.
На вкладке *Проверка и создание.
Просмотрите ресурсы, которые были автоматически созданы для вас, и выберите Создать.
Подождите, пока развертывание завершится. Если развертывание завершилось неудачно, выберите Сведения об операции рядом с этапом сбоя, чтобы получить дополнительную информацию о причине сбоя. Выберите Повторное развертывание, чтобы снова попытаться развернуть ресурсы. Вы можете изменить параметры перед развертыванием.

На портале Azure перейдите на страницу учетной записи хранения. В меню слева выберите События.
На главной панели перейдите на вкладку Azure Data Explorer.

Откроется панель Подключение к данным с выбранной вкладкой Основные сведения.

На вкладке Основное заполните форму, указав следующую информацию:

Снимок экрана: вкладка

Параметр	Рекомендуемое значение	Описание поля
Имя подключения к данным	test-grid-connection	Имя создаваемого подключения в Azure Data Explorer.
Storage account	gridteststorage1	Учетная запись хранения, в которой был получен доступ к мастеру. Заполняется автоматически.
Тип события	Созданный BLOB-объект или переименованный BLOB-объект	Тип события, запускающего загрузку. Переименование BLOB-объектов поддерживается только для хранилища ADLSv2. Чтобы переименовать большой двоичный объект, перейдите к большому двоичному объекту в портал Azure, щелкните его правой кнопкой мыши и выберите "Переименовать". Поддерживаемые типы: Microsoft.Storage.BlobCreated или Microsoft.Storage.BlobRenamed.
Создание ресурсов	Автоматически	Определите, хотите ли вы, чтобы Azure Data Explorer создавал для вас подписку на Сетку событий, пространство имен Центров событий и сами Центры событий. Чтобы создать ресурсы вручную, см. раздел Создание ресурсов для приема Сетки событий вручную

Выберите Далее> Свойства приема.
На вкладке Свойства приема.

Заполните форму, указав следующую информацию. Имена таблиц и отображений чувствительны к регистру:

Снимок экрана: вкладка

Параметр	Рекомендуемое значение	Описание поля
Отток подписок		Ваша подписка на Azure Data Explorer.
Имя кластера	TestCluster	Имя кластера для приема данных.
Имя базы данных	TestDatabase	Целевая база данных, созданная в кластере TestCluster.
Разрешить маршрутизацию данных в другие базы данных (подключение данных к нескольким базам данных)	Не разрешать	Включите этот параметр, если требуется переопределить целевую базу данных по умолчанию, связанную с подключением к данным. Дополнительные сведения о маршрутизации в базы данных см. в разделе Маршрутизация событий.
Имя таблицы	TestTable	Целевая таблица, созданная в базе данных TestDatabase.
Формат данных	JSON	Поддерживаемые форматы: Avro, CSV, JSON, MULTILINE JSON, ORC, PARQUET, PSV, SCSV, SOHSV, TSV, TXT, TSVE, APACHEAVRO, RAW и W3CLOG. Поддерживаются форматы сжатия ZIP и GZIP.
Сопоставление	TestMapping	Сопоставление, созданное в TestDatabase, которое сопоставляет входящие данные с именами столбцов и типами данных TestTable. Если значение не указано, используется сопоставление данных удостоверений, полученное из схемы таблицы.
Расширенные настройки	У моих данных есть заголовки	Игнорирует заголовки. Поддерживается для файлов типа *SV.

Примечание.

Вам не нужно указывать все параметры маршрутизации данных. Принимаются также частично заполненные параметры.

Выберите Далее: Проверка и создание.
На вкладке Проверка и создание.
Просмотрите ресурсы, которые были автоматически созданы для вас, и выберите Создать.
Подождите, пока развертывание завершится. Если развертывание завершилось неудачно, выберите Сведения об операции рядом с этапом сбоя, чтобы получить дополнительную информацию о причине сбоя. Выберите Повторное развертывание, чтобы снова попытаться развернуть ресурсы. Вы можете изменить параметры перед развертыванием.

Установите пакет NuGet Microsoft.Azure.Management.Kusto.
Создайте субъект приложения Azure AD для проверки подлинности. Вам потребуется идентификатор каталога (клиента), идентификатор приложения и секрет клиента.

Выполните следующий код.

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 eventGridConnectionName = "myeventgridconnect";
//The event hub and storage account that are created as part of the Prerequisites
var eventHubResourceId = "/subscriptions/<storageAccountSubscriptionId>/resourceGroups/<storageAccountResourceGroupName>/providers/Microsoft.Storage/storageAccounts/<storageAccountName>";
var storageAccountResourceId = "/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 mappingRuleName = "StormEvents_CSV_Mapping";
var dataFormat = DataFormat.CSV;
var blobStorageEventType = "Microsoft.Storage.BlobCreated";
var databaseRouting = "Multi";
var eventGridConnectionData = new EventGridDataConnection(
    storageAccountResourceId, eventHubResourceId, consumerGroup,
    tableName: tableName, location: location, mappingRuleName: mappingRuleName,
    dataFormat: dataFormat, blobStorageEventType: blobStorageEventType, databaseRouting: databaseRouting
);
await kustoManagementClient.DataConnections.CreateOrUpdateAsync(
    resourceGroupName, clusterName, databaseName, eventGridConnectionName, eventGridConnectionData
);

Параметр	Рекомендуемое значение	Описание поля
tenantId	xxxxxxxx-xxxxx-xxxx-xxxx-xxxxxxxxx	Идентификатор клиента. Также известен как идентификатор каталога.
subscriptionId	xxxxxxxx-xxxxx-xxxx-xxxx-xxxxxxxxx	Идентификатор подписки, используемой для создания ресурсов.
clientId	xxxxxxxx-xxxxx-xxxx-xxxx-xxxxxxxxx	Идентификатор клиента приложения, которое имеет доступ к ресурсам в клиенте.
clientSecret	PlaceholderClientSecret	Секрет клиента приложения, которое имеет доступ к ресурсам в клиенте.
resourceGroupName	testrg	Имя группы ресурсов, содержащей ваш кластер.
clusterName	mykustocluster	Имя вашего кластера.
databaseName	mykustodatabase	Имя целевой базы данных в вашем кластере.
eventGridConnectionName	myeventgridconnect	Необходимое имя вашего подключения для передачи данных.
tableName	StormEvents	Имя целевой таблицы в целевой базе данных.
mappingRuleName	StormEvents_CSV_Mapping	Имя сопоставления столбцов, связанного с целевой таблицей.
dataFormat	csv	Формат данных сообщения.
eventHubResourceId	Идентификатор ресурса	Идентификатор ресурса вашего концентратора событий, в котором Сетка событий настроена для отправки событий.
storageAccountResourceId	Идентификатор ресурса	Идентификатор ресурса вашей учетной записи хранения, в которой хранятся данные для приема.
consumerGroup	$Default	Группа потребителей вашего концентратора событий.
расположение	Центральная часть США	Расположение ресурса подключения к данным.
blobStorageEventType	Microsoft.Storage.BlobCreated	Тип события, запускающего загрузку. Поддерживаемые события: Microsoft.Storage.BlobCreated или Microsoft.Storage.BlobRenamed. Переименование BLOB-объектов поддерживается только для хранилища ADLSv2.
databaseRouting	Multi или Single	Маршрутизация базы данных для подключения. Если задано значение Single, подключение к данным будет направляться в отдельную базу данных в кластере, как указано в параметре databaseName. Если задать значение *Multi, можно переопределить целевую базу данных по умолчанию с помощью свойства приема базы данных.* Дополнительные сведения см. в разделе Маршрутизация событий.

Импортируйте необходимые библиотеки.
```
pip install azure-common
pip install azure-mgmt-kusto
```
Создайте субъект приложения Azure AD для проверки подлинности. Вам потребуется идентификатор каталога (клиента), идентификатор приложения и секрет клиента.

Выполните следующий код.

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

Параметр	Рекомендуемое значение	Описание поля
tenant_id	xxxxxxxx-xxxxx-xxxx-xxxx-xxxxxxxxx	Идентификатор клиента. Также известен как идентификатор каталога.
subscription_id	xxxxxxxx-xxxxx-xxxx-xxxx-xxxxxxxxx	Идентификатор подписки, используемой для создания ресурсов.
client_id	xxxxxxxx-xxxxx-xxxx-xxxx-xxxxxxxxx	Идентификатор клиента приложения, которое имеет доступ к ресурсам в клиенте.
client_secret	xxxxxxxxxxxxxx	Секрет клиента приложения, которое имеет доступ к ресурсам в клиенте.
resource_group_name	testrg	Имя группы ресурсов, содержащей ваш кластер.
cluster_name	mykustocluster	Имя вашего кластера.
database_name	mykustodatabase	Имя целевой базы данных в вашем кластере.
data_connection_name	myeventhubconnect	Необходимое имя вашего подключения для передачи данных.
table_name	StormEvents	Имя целевой таблицы в целевой базе данных.
mapping_rule_name	StormEvents_CSV_Mapping	Имя сопоставления столбцов, связанного с целевой таблицей.
database_routing	Multi или Single	Маршрутизация базы данных для подключения. Если задано значение Single, подключение к данным будет направляться в отдельную базу данных в кластере, как указано в параметре databaseName. Если задать значение *Multi, можно переопределить целевую базу данных по умолчанию с помощью свойства приема базы данных.* Дополнительные сведения см. в разделе Маршрутизация событий.
data_format	csv	Формат данных сообщения.
event_hub_resource_id	Идентификатор ресурса	Идентификатор ресурса вашего концентратора событий, в котором Сетка событий настроена для отправки событий.
storage_account_resource_id	Идентификатор ресурса	Идентификатор ресурса вашей учетной записи хранения, в которой хранятся данные для приема.
consumer_group	$Default	Группа потребителей вашего концентратора событий.
расположение	Центральная часть США	Расположение ресурса подключения к данным.
blob_storage_event_type	Microsoft.Storage.BlobCreated	Тип события, запускающего загрузку. Поддерживаемые события: Microsoft.Storage.BlobCreated или Microsoft.Storage.BlobRenamed. Переименование BLOB-объектов поддерживается только для хранилища ADLSv2.

В следующем примере показан шаблон Azure Resource Manager для добавления подключения к данным Сетки событий. Вы можете изменить и развернуть шаблон на портале Azure с помощью формы.

{
    "$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 hub 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."
            }
        },
        "StorageAccounts_storagedemo_name": {
            "type": "string",
            "defaultValue": "storagedemo",
            "metadata": {
                "description": "Specifies the storage account name"
            }
        },
        "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 resources"
            }
        },
        "resourceGroup": {
            "type": "string",
            "defaultValue": "[resourceGroup().name]",
            "metadata": {
                "description": "Specifies the resourceGroup of the resources"
            }
        },
        "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": "EventGrid",
            "properties": {
                "managedIdentityResourceId": "[resourceId('Microsoft.Kusto/clusters', parameters('clusters_kustocluster_name'))]",
                "storageAccountResourceId": "[resourceId(parameters('subscriptionId'), parameters('resourceGroup'), 'Microsoft.Storage/storageAccounts', parameters('StorageAccounts_storagedemo_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')]"
            }
        }
    ]
}

Удаление подключения к данным сетки событий

Чтобы удалить подключение сетки событий из портал Azure, сделайте следующее:

Перейдите в кластер. В меню слева выберите "Базы данных". Затем выберите базу данных, содержащую целевую таблицу.
В меню слева выберите подключения к данным. Затем установите флажок рядом с соответствующим подключением к данным сетки событий.
В верхней строке меню нажмите кнопку "Удалить".

Чтобы удалить подключение сетки событий, выполните следующую команду:

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

Чтобы удалить подключение сетки событий, выполните следующую команду:

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)

Следующие шаги

Обработка данных из концентратора событий с помощью Azure Stream Analytics

Поделиться через

Создание подключения к данным сетки событий для Azure Data Explorer

Необходимые компоненты

Создание подключения к данным в Сетке событий

Удаление подключения к данным сетки событий

Следующие шаги

Дополнительные ресурсы