为 Azure 数据资源管理器创建事件网格数据连接

本文介绍如何使用事件网格数据连接将 blob 从存储帐户引入到 Azure 数据资源管理器。 你将创建一个用于设置 Azure 事件网格订阅的事件网格数据连接。 事件网格订阅通过 Azure 事件中心将事件从存储帐户路由到 Azure 数据资源管理器。

有关如何从事件网格引入 Azure 数据资源管理器的一般信息，请参阅连接到事件网格。

注意

若要使用事件网格连接实现最佳性能，请通过 Blob 元数据设置 rawSizeBytes 引入属性。 有关详细信息，请参阅引入属性。

先决条件

• Azure 订阅。 创建免费 Azure 帐户。
• Azure 数据资源管理器群集和数据库。 创建群集和数据库。
• 一个目标表。 创建表或使用现有表。
• 表的引入映射。
• 一个存储帐户。 可以在 BlobStorage、StorageV2 或 Data Lake Storage Gen2 的 Azure 存储帐户上设置事件网格通知订阅。

创建事件网格数据连接

在本部分，我们将在事件网格与 Azure 数据资源管理器表之间建立连接。

门户 - Azure 数据资源管理器

1. 在创建的群集下，选择“数据库”“TestDatabase”。

   

2. 选择“数据引入”“添加数据连接”。

   

3. 在“基本信息”下，选择连接类型：Blob 存储，然后使用以下信息填写表单：

   

   设置	建议的值	字段说明
   数据连接名称	test-grid-connection	要在 Azure 数据资源管理器中创建的连接的名称。
   存储帐户订阅	订阅 ID	存储帐户所在的订阅 ID。
   存储帐户	gridteststorage1	前面创建的存储帐户的名称。
   事件类型	“已创建 Blob”或“已重命名 Blob”	触发引入的事件类型。 “已重命名 Blob”仅支持 ADLSv2 存储。 若要重命名 blob，请导航到 Azure 门户中的 blob，右键单击该 blob 并选择“重命名”。 支持的类型为：Microsoft.Storage.BlobCreated 或 Microsoft.Storage.BlobRenamed。
   资源创建	自动	定义是否希望 Azure 数据资源管理器为你创建事件网格订阅、事件中心命名空间和事件中心。 若要手动创建资源，请参阅手动创建资源以实现事件网格引入

4. 如果要跟踪特定主题，请选择“筛选器设置”。 按如下所述设置通知筛选器：

   • 前缀字段是主题的文本前缀。 由于应用的模式是 startswith，因此可以跨越多个容器、文件夹或 blob 应用。 不允许通配符。
     • 若要在 Blob 容器上定义筛选器，必须按如下所示设置字段：。
     • 若要在 Blob 前缀（或 Azure Data Lake Gen2 中的文件夹）上定义筛选器，必须按如下所示设置字段：。
   • “后缀”字段是 Blob 的文本后缀。 不允许通配符。
   • “区分大小写”字段指示前缀和后缀筛选器是否区分大小写
   • 有关筛选事件的详细信息，请参阅 blob 存储事件。

   

5. 选择“下一步：引入属性”。

6. 使用以下信息填写窗体。 表和映射名称区分大小写：

   

   引入属性：

   设置	建议的值	字段说明
   允许将数据路由到其他数据库(多数据库数据连接)	不允许	若要重写与数据连接关联的默认目标数据库，请启用此选项。 有关数据库路由的详细信息，请参阅事件路由。
   表名	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 类型文件。

   注意

   无需指定所有默认路由设置。 部分设置也是接受的。

7. 选择“下一步: 查看 + 创建”

8. 在*“查看 + 创建”下面。

9. 查看自动创建的资源，并选择“创建”。

   

10. 等待部署完成。 如果部署失败，请在失败阶段的旁边选择“操作详细信息”，获取有关失败原因的详细信息。 选择“重新部署”，再次尝试部署资源。 你可以在部署之前修改参数。

    

门户 - Azure 存储

1. 浏览到 Azure 门户中的存储帐户。 在左侧菜单中，选择“事件”。

2. 在主窗格中，选择“Azure 数据资源管理器”选项卡。

   

“数据连接”窗格处于打开状态，并选中了“基本信息”选项卡 。

1. 在“基本信息”下面，使用以下信息填写表单：

   

   设置	建议的值	字段说明
   数据连接名称	test-grid-connection	要在 Azure 数据资源管理器中创建的连接的名称。
   存储帐户	gridteststorage1	用于访问此向导的存储帐户。 自动填充。
   事件类型	“已创建 Blob”或“已重命名 Blob”	触发引入的事件类型。 “已重命名 Blob”仅支持 ADLSv2 存储。 若要重命名 blob，请导航到 Azure 门户中的 blob，右键单击该 blob 并选择“重命名”。 支持的类型为：Microsoft.Storage.BlobCreated 或 Microsoft.Storage.BlobRenamed。
   资源创建	自动	定义是否希望 Azure 数据资源管理器为你创建事件网格订阅、事件中心命名空间和事件中心。 若要手动创建资源，请参阅手动创建资源以实现事件网格引入

2. 选择“下一步> 引入属性”。

3. 在“引入属性”下面。

4. 使用以下信息填写窗体。 表和映射名称区分大小写：

   

   设置	建议的值	字段说明
   订阅		你的 Azure 数据资源管理器订阅。
   群集名称	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 类型文件。

   注意

   无需指定所有数据路由设置。 部分设置也是接受的。

5. 选择“下一步: 查看 + 创建”

6. 在“查看 + 创建”下面。

7. 查看自动创建的资源，并选择“创建”。

   

8. 等待部署完成。 如果部署失败，请在失败阶段的旁边选择“操作详细信息”，获取有关失败原因的详细信息。 选择“重新部署”，再次尝试部署资源。 你可以在部署之前修改参数。

   

C#

1. 安装 Microsoft.Azure.Management.Kusto NuGet 包。

2. 创建用于身份验证的 Azure AD 应用程序主体。 需要目录（租户）ID、应用程序 ID 和客户端机密。

3. 运行以下代码。

   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	租户 ID。 也称为目录 ID。
   subscriptionId	xxxxxxxx-xxxxx-xxxx-xxxx-xxxxxxxxx	用于创建资源的订阅 ID。
   clientId	xxxxxxxx-xxxxx-xxxx-xxxx-xxxxxxxxx	可以访问租户中资源的应用程序的客户端 ID。
   clientSecret	PlaceholderClientSecret	可以访问租户中资源的应用程序的客户端密码。
   resourceGroupName	testrg	包含群集的资源组的名称。
   clusterName	mykustocluster	群集的名称。
   databaseName	mykustodatabase	群集中目标数据库的名称。
   eventGridConnectionName	myeventgridconnect	所需的数据连接名称。
   tableName	StormEvents	目标数据库中目标表的名称。
   mappingRuleName	StormEvents_CSV_Mapping	与目标表相关的列映射的名称。
   dataFormat	csv	消息的数据格式。
   eventHubResourceId	资源 ID	将事件网格配置为发送事件的事件中心的资源 ID。
   storageAccountResourceId	资源 ID	包含要引入数据的存储帐户的资源 ID。
   consumerGroup	$Default	事件中心的使用者组。
   location	美国中部	数据连接资源的位置。
   blobStorageEventType	Microsoft.Storage.BlobCreated	触发引入的事件类型。 支持的事件为：Microsoft.Storage.BlobCreated 或 Microsoft.Storage.BlobRenamed。 仅 ADLSv2 存储支持 Blob 重命名。
   databaseRouting	多或单	连接的数据库路由。 如果将此值设置为“单”，数据连接将按“databaseName”设置中指定的那样路由到群集中的单个数据库。 如果将此值设置为“多”，可使用数据库引入属性重写默认目标数据库。 有关详细信息，请参阅事件路由。

Python

1. 安装所需的库。

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

2. 创建用于身份验证的 Azure AD 应用程序主体。 需要目录（租户）ID、应用程序 ID 和客户端机密。

3. 运行以下代码。

   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	租户 ID。 也称为目录 ID。
   subscription_id	xxxxxxxx-xxxxx-xxxx-xxxx-xxxxxxxxx	用于创建资源的订阅 ID。
   client_id	xxxxxxxx-xxxxx-xxxx-xxxx-xxxxxxxxx	可以访问租户中资源的应用程序的客户端 ID。
   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	多或单	连接的数据库路由。 如果将此值设置为“单”，数据连接将按“databaseName”设置中指定的那样路由到群集中的单个数据库。 如果将此值设置为“多”，可使用数据库引入属性重写默认目标数据库。 有关详细信息，请参阅事件路由。
   data_format	csv	消息的数据格式。
   event_hub_resource_id	资源 ID	将事件网格配置为发送事件的事件中心的资源 ID。
   storage_account_resource_id	资源 ID	包含要引入的数据的存储帐户的资源 ID。
   consumer_group	$Default	事件中心的使用者组。
   location	美国中部	数据连接资源的位置。
   blob_storage_event_type	Microsoft.Storage.BlobCreated	触发引入的事件类型。 支持的事件为：Microsoft.Storage.BlobCreated 或 Microsoft.Storage.BlobRenamed。 仅 ADLSv2 存储支持 Blob 重命名。

ARM 模板

以下示例显示用于添加事件网格数据连接的 Azure 资源管理器模板。 可以使用此窗体在 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')]"
            }
        }
    ]
}

删除事件网格数据连接

Portal

若要从 Azure 门户中删除事件网格连接，请执行以下操作：

1. 转到你的群集。 在左侧菜单中选择“数据库”。 然后选择包含目标表的数据库。
2. 从左侧菜单选择“数据连接”。 然后，选中相关的事件网格数据连接旁边的复选框。
3. 从顶部菜单栏中选择“删除”。

C#

若要删除事件网格连接，请运行以下命令：

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

Python

若要删除事件网格连接，请运行以下命令：

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 流分析处理事件中心的数据