建立 Azure 數據總管的事件方格數據連線

在本文中，您將瞭解如何使用事件方格數據連線，將記憶體帳戶中的 Blob 內嵌至 Azure 數據總管。 您將建立設定 Azure 事件方格 訂用帳戶的事件方格資料連線。 事件方格訂用帳戶會透過 Azure 事件中樞，將事件從記憶體帳戶路由傳送至 Azure 資料總管。

如需從事件方格擷取至 Azure 數據總管的一般資訊，請參閱 連線至事件方格。

注意

若要使用事件方格聯機達到最佳效能，請透過 Blob 元數據設定 rawSizeBytes 擷取屬性。 如需詳細資訊，請參閱 擷取屬性。

必要條件

• Azure 訂用帳戶。 建立免費的 Azure 帳戶。
• Azure 資料總管叢集和資料庫。 建立叢集和資料庫。
• 目的地數據表。 建立數據表 或使用現有的數據表。
• 數據表的擷取對應。
• 記憶體帳戶。 事件方格通知訂用帳戶可以在、 StorageV2或 Data Lake Storage Gen2 的 Azure 儲存體 帳戶BlobStorage上設定。

建立事件方格資料連線

在本節中，您將建立事件方格與 Azure 數據總管數據表之間的連線。

入口網站 - Azure 數據總管

1. 在您建立的叢集下，選取 [資料庫>TestDatabase]。

   

2. 選取 [數據擷取>][新增數據連線]。

   

3. 在 [基本] 底下，選取連線類型：Blob 記憶體，然後填寫窗體，並提供下列資訊：

   

   設定	建議的值	欄位描述
   資料連線名稱	test-grid-connection	您想要在 Azure 資料總管中建立的連接名稱。
   儲存體帳戶訂用帳戶	您的訂用帳戶識別碼	儲存體帳戶所在的訂用帳戶識別碼。
   儲存體帳戶	gridteststorage1	您先前建立之儲存體帳戶的名稱。
   事件類型	blob 已建立 或 blob 已重新命名	觸發擷取的事件類別。 只有 ADLSv2 儲存體支援 blob 已重新命名。 若要重新命名 Blob，請流覽至 Azure 入口網站 中的 Blob，以滑鼠右鍵按兩下 Blob，然後選取 [重新命名]。 支援的型別：Microsoft.Storage.BlobCreated 或 Microsoft.Storage.BlobRenamed。
   資源建立	自動	定義您是否要讓 Azure 數據總管為您建立事件方格訂用帳戶、事件中樞命名空間和事件中樞。 若要手動建立資源，請參閱 手動建立事件方格擷取的資源

4. 若要追蹤特定的主題，請選取 [篩選條件設定]。 設定通知的篩選條件，如下所示：

   • 前置詞欄位是主題的常值前置詞。 套用的模式是 startswith時可跨越多個容器、資料夾或 blob。 不允許使用萬用字元。
     • 若要在 blob 容器上定義篩選條件，欄位必須按照以下方式設定：/blobServices/default/containers/[container prefix]。
     • 若要在 blob 前置詞 (或 Azure Data Lake Gen2) 上定義資料夾，欄位必須按照以下方式設定：/blobServices/default/containers/[container name]/blobs/[folder/blob prefix]。
   • 後置詞欄位是 blob 的常值後置詞。 不允許使用萬用字元。
   • 區分大小寫欄位顯示前置詞和後置詞篩選條件是否區分大小寫
   • 如需篩選事件的詳細資訊，請參考 blob 儲存體事件。

   

5. 選取 [下一步：內嵌屬性]。

6. 在表單中填寫以下資訊。 資料表和對應名稱需區分大小寫：

   

   內嵌屬性：

   設定	建議的值	欄位描述
   允許將資料路由傳送至其他資料庫（多資料庫資料連線）	不允許	如果您想要覆寫與數據連線相關聯的默認目標資料庫，請開啟此選項。 如需資料庫路由的詳細資訊，請參閱 事件路由。
   資料表名稱	TestTable	您在 TestDatabase 中建立的資料表。
   資料格式	JSON	支援的格式為 Avro、CSV、JSON、MULTILINE JSON、ORC、PARQUET、PSV、SCSV、SOHSV、TSV、TXT、TSVE、APACHEAVRO 和 W3CLOG。 支援的壓縮選項為 Zip 和 Gzip。
   對應	TestMapping	您在 TestDatabase 中建立的對應，會將傳入的數據對應至 TestTable 的數據行名稱和數據類型。 如果未指定， 則會使用衍生自數據表架構的識別數據對應 。
   進階設定	我的資料有標頭	忽略標頭。 支援 *SV 類型的檔案。

   注意

   您不必指定所有預設路由設定。 也會接受部分設定。

7. 選取 [下一步：檢閱 + 建立]

8. 在 [檢閱 + 建立] 底下。

9. 檢閱自動建立的資源，並選取 [建立]。

   

10. 請等候部署完成。 如果您的部署失敗，請選取失敗階段旁的作業詳細資訊，了解失敗原因的詳細資訊。 選取 [重新部署] 嘗試再次部署資源。 部署前，您可變更參數。

    

入口網站 - Azure 記憶體

1. 流覽至 Azure 入口網站 中的記憶體帳戶。 在左側功能表上，選取 [事件]

2. 在主窗格中，選取 [ Azure 數據總管] 索引標籤 。

   

[數據連線] 窗格隨即開啟，並已選取 [基本] 索引標籤。

1. 在 [ 基本概念] 底下，以下列資訊填寫窗體：

   

   設定	建議的值	欄位描述
   資料連線名稱	test-grid-connection	您想要在 Azure 資料總管中建立的連接名稱。
   儲存體帳戶	gridteststorage1	您從中存取此精靈的記憶體帳戶。 自動填入。
   事件類型	blob 已建立 或 blob 已重新命名	觸發擷取的事件類別。 只有 ADLSv2 儲存體支援 blob 已重新命名。 若要重新命名 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 和 W3CLOG。 支援的壓縮選項為 Zip 和 Gzip。
   對應	TestMapping	您在 TestDatabase 中建立的對應，會將傳入的數據對應至 TestTable 的數據行名稱和數據類型。 如果未指定， 則會使用衍生自數據表架構的識別數據對應 。
   進階設定	我的資料有標頭	忽略標頭。 支援 *SV 類型的檔案。

   注意

   您不需要指定所有資料 路由設定。 也會接受部分設定。

5. 選取 [下一步：檢閱 + 建立]

6. 在 [檢閱 + 建立] 底下。

7. 檢閱自動建立的資源，並選取 [建立]。

   

8. 請等候部署完成。 如果您的部署失敗，請選取失敗階段旁的作業詳細資訊，了解失敗原因的詳細資訊。 選取 [重新部署] 嘗試再次部署資源。 部署前，您可變更參數。

   

C#

1. 安裝 Microsoft.Azure.Management.Kusto NuGet 套件。

2. 建立要用於驗證的 Azure AD 應用程式主體 。 您將需要目錄（租使用者）識別碼、應用程式識別碼和客戶端密碼。

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	您的租用戶識別碼。 也稱為目錄標識碼。
   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	事件中樞的取用者群組。
   location	Central US	數據連線資源的位置。
   blobStorageEventType	Microsoft.Storage.BlobCreated	觸發擷取的事件類別。 支援的事件包括：Microsoft.Storage.BlobCreated 或 Microsoft.Storage.BlobRenamed。 只有ADLSv2記憶體才支援 Blob 重新命名。
   databaseRouting	多重 或 單一	線上的資料庫路由。 如果您將值設定為Single，數據連線將會路由傳送至叢集中的單一資料庫，如databaseName設定中所指定。 如果您將值設定為 Multi，您可以使用資料庫擷取屬性覆寫預設目標資料庫。 如需詳細資訊，請參閱 事件路由。

Python

1. 安裝必要的程式庫。

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

2. 建立要用於驗證的 Azure AD 應用程式主體 。 您將需要目錄（租使用者）識別碼、應用程式識別碼和客戶端密碼。

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	您的租用戶識別碼。 也稱為目錄標識碼。
   subscription_id	xxxxxxxx-xxxxx-xxxx-xxxx-xxxxxxxxx	您用來建立資源的訂用帳戶標識碼。
   client_id	xxxxxxxx-xxxxx-xxxx-xxxx-xxxxxxxxx	應用程式可存取租用戶中資源的用戶端標識碼。
   client_secret	xxxxxxxx	應用程式可存取租用戶中資源的客戶端密碼。
   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	多重 或 單一	線上的資料庫路由。 如果您將值設定為Single，數據連線將會路由傳送至叢集中的單一資料庫，如databaseName設定中所指定。 如果您將值設定為 Multi，您可以使用資料庫擷取屬性覆寫預設目標資料庫。 如需詳細資訊，請參閱 事件路由。
   data_format	csv	訊息的數據格式。
   event_hub_resource_id	資源識別碼	事件方格設定為傳送事件的事件中樞資源標識符。
   storage_account_resource_id	資源識別碼	記憶體帳戶的資源標識碼，此帳戶會保存要擷取的數據。
   consumer_group	$Default	事件中樞的取用者群組。
   location	Central US	數據連線資源的位置。
   blob_storage_event_type	Microsoft.Storage.BlobCreated	觸發擷取的事件類別。 支援的事件包括：Microsoft.Storage.BlobCreated 或 Microsoft.Storage.BlobRenamed。 只有ADLSv2記憶體才支援 Blob 重新命名。

ARM 範本

下列範例示範用於新增事件方格數據連線的 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 入口網站 移除事件方格連線，請執行下列動作：

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 串流分析處理來自事件中樞的數據