使用事件中心和 .NET 发送和接收 Atlas Kafka 主题消息

备注

Microsoft Purview 数据目录将其名称更改为 Microsoft Purview 统一目录。 所有功能将保持不变。 当新的 Microsoft Purview 数据治理体验在你的区域中正式发布时,你将看到名称更改。 检查你所在的区域的名称

本快速入门介绍如何发送和接收 Atlas Kafka 主题事件。 我们将使用 Azure 事件中心Azure.Messaging.EventHubs .NET 库。

先决条件

如果不熟悉事件中心,请在完成本快速入门之前参阅 事件中心概述

若要遵循本快速入门,需要满足某些先决条件:

将消息发布到 Microsoft Purview

让我们创建一个 .NET Core 控制台应用程序,该应用程序通过事件中心 Kafka 主题 (ATLAS_HOOK)将事件发送到 Microsoft Purview。

若要将消息发布到 Microsoft Purview,需要 一个托管事件中心,或 至少一个具有挂钩配置的事件中心

创建 Visual Studio 项目

接下来,在 Visual Studio 中创建 C# .NET 控制台应用程序:

  1. 启动 Visual Studio
  2. 在“开始”窗口中,选择“新建项目>控制台应用 (.NET Framework) ”。 需要 .NET 版本 4.5.2 或更高版本。
  3. “项目名称”中,输入 PurviewKafkaProducer
  4. 选择“ 创建 ”以创建项目。

创建控制台应用程序

  1. 启动 Visual Studio 2022。
  2. 选择 “创建新项目”。
  3. 在“ 创建新项目 ”对话框中,执行以下步骤:如果未看到此对话框,请在菜单中选择“ 文件 ”,选择“ 新建”,然后选择“ 项目”。
    1. 对于编程语言,请选择 “C# ”。
    2. 为应用程序类型选择 “控制台 ”。
    3. 从结果列表中选择“ 控制台应用” (.NET“核心) ”。
    4. 然后,选择“下一步”

添加事件中心 NuGet 包

  1. 从菜单中选择“ 工具>”“NuGet 包管理器>包管理器控制台 ”。

  2. 运行以下命令以安装 Azure.Messaging.EventHubs NuGet 包和 Azure.Messaging.EventHubs.Producer NuGet 包:

    Install-Package Azure.Messaging.EventHubs
    
    Install-Package Azure.Messaging.EventHubs.Producer
    

编写将消息发送到事件中心的代码

  1. 将以下 using 语句添加到 Program.cs 文件的顶部:

    using System;
    using System.Text;
    using System.Threading.Tasks;
    using Azure.Messaging.EventHubs;
    using Azure.Messaging.EventHubs.Producer;
    
  2. 为事件中心连接字符串和事件中心名称向 类添加常量Program

    private const string connectionString = "<EVENT HUBS NAMESPACE - CONNECTION STRING>";
    private const string eventHubName = "<EVENT HUB NAME>";
    
  3. Main将 方法替换为以下async Main方法,并添加 以async ProduceMessage将消息推送到 Microsoft Purview 中。 有关详细信息,请参阅代码中的注释。

        static async Task Main()
        {
            // Read from the default consumer group: $Default
            string consumerGroup = EventHubConsumerClient.DefaultConsumerGroupName;
    
     		/ Create an event producer client to add events in the event hub
            EventHubProducerClient producer = new EventHubProducerClient(ehubNamespaceConnectionString, eventHubName);
    
     		await ProduceMessage(producer);
        }
    
     	static async Task ProduceMessage(EventHubProducerClient producer)
    
        {
     		// Create a batch of events 
     		using EventDataBatch eventBatch = await producerClient.CreateBatchAsync();
    
     		// Add events to the batch. An event is a represented by a collection of bytes and metadata. 
     		eventBatch.TryAdd(new EventData(Encoding.UTF8.GetBytes("<First event>")));
     		eventBatch.TryAdd(new EventData(Encoding.UTF8.GetBytes("<Second event>")));
     		eventBatch.TryAdd(new EventData(Encoding.UTF8.GetBytes("<Third event>")));
    
     		// Use the producer client to send the batch of events to the event hub
     		await producerClient.SendAsync(eventBatch);
     		Console.WriteLine("A batch of 3 events has been published.");
    
     	}
    
  4. 生成项目。 确保没有任何错误。

  5. 运行程序并等待确认消息。

    备注

    有关包含详细信息注释的完整源代码,请参阅 GitHub 中的此文件

使用创建实体 JSON 消息创建包含两列的 sql 表的示例代码

	
	{
    "msgCreatedBy":"nayenama",
    "message":{
        "type":"ENTITY_CREATE_V2",
        "user":"admin",
        "entities":{
            "entities":[
                {
                    "typeName":"azure_sql_table",
                    "attributes":{
                        "owner":"admin",
                        "temporary":false,
                        "qualifiedName":"mssql://nayenamakafka.eventhub.sql.net/salespool/dbo/SalesOrderTable",
                        "name":"SalesOrderTable",
                        "description":"Sales Order Table added via Kafka"
                    },
                    "relationshipAttributes":{
                        "columns":[
                            {
                                "guid":"-1102395743156037",
                                "typeName":"azure_sql_column",
                                "uniqueAttributes":{
                                    "qualifiedName":"mssql://nayenamakafka.eventhub.sql.net/salespool/dbo/SalesOrderTable#OrderID"
                                }
                            },
                            {
                                "guid":"-1102395743156038",
                                "typeName":"azure_sql_column",
                                "uniqueAttributes":{
                                    "qualifiedName":"mssql://nayenamakafka.eventhub.sql.net/salespool/dbo/SalesOrderTable#OrderDate"
                                }
                            }
                        ]
                    },
                    "guid":"-1102395743156036",
                    "version":0
                }
            ],
            "referredEntities":{
                "-1102395743156037":{
                    "typeName":"azure_sql_column",
                    "attributes":{
                        "owner":null,
                        "userTypeId":61,
                        "qualifiedName":"mssql://nayenamakafka.eventhub.sql.net/salespool/dbo/SalesOrderTable#OrderID",
                        "precision":23,
                        "length":8,
                        "description":"Sales Order ID",
                        "scale":3,
                        "name":"OrderID",
                        "data_type":"int"
                    },
                    "relationshipAttributes":{
                        "table":{
                            "guid":"-1102395743156036",
                            "typeName":"azure_sql_table",
                            "entityStatus":"ACTIVE",
                            "displayText":"SalesOrderTable",
                            "uniqueAttributes":{
                                "qualifiedName":"mssql://nayenamakafka.eventhub.sql.net/salespool/dbo/SalesOrderTable"
                            }
                        }
                    },
                    "guid":"-1102395743156037",
                    "version":2
                },
                "-1102395743156038":{
                    "typeName":"azure_sql_column",
                    "attributes":{
                        "owner":null,
                        "userTypeId":61,
                        "qualifiedName":"mssql://nayenamakafka.eventhub.sql.net/salespool/dbo/SalesOrderTable#OrderDate",
                        "description":"Sales Order Date",
                        "scale":3,
                        "name":"OrderDate",
                        "data_type":"datetime"
                    },
                    "relationshipAttributes":{
                        "table":{
                            "guid":"-1102395743156036",
                            "typeName":"azure_sql_table",
                            "entityStatus":"ACTIVE",
                            "displayText":"SalesOrderTable",
                            "uniqueAttributes":{
                                "qualifiedName":"mssql://nayenamakafka.eventhub.sql.net/salespool/dbo/SalesOrderTable"
                            }
                        }
                    },
                    "guid":"-1102395743156038",
                    "status":"ACTIVE",
                    "createdBy":"ServiceAdmin",
                    "version":0
                }
            }
        }
    },
    "version":{
        "version":"1.0.0"
    },
    "msgCompressionKind":"NONE",
    "msgSplitIdx":1,
    "msgSplitCount":1
}


接收 Microsoft Purview 消息

接下来,了解如何编写使用事件处理程序从事件中心接收消息的 .NET Core 控制台应用程序。 事件处理程序管理来自事件中心的持久检查点和并行接收。 这简化了接收事件的过程。 需要使用 ATLAS_ENTITIES 事件中心从 Microsoft Purview 接收消息。

若要从 Microsoft Purview 接收消息,需要 托管事件中心事件中心通知配置

警告

事件中心 SDK 使用最新版本的可用存储 API。 该版本不一定在 Stack Hub 平台上可用。 如果在 Azure Stack Hub 上运行此代码,除非以使用的特定版本为目标,否则会遇到运行时错误。 如果使用 Azure Blob 存储 作为检查点存储,请查看 Azure Stack Hub 内部版本的受支持的 Azure 存储 API 版本,并在代码中以该版本为目标。

存储服务的最高可用版本是版本 2019-02-02。 默认情况下,事件中心 SDK 客户端库在发布 SDK) 时使用 Azure (2019-07-07 上可用的最高版本。 如果使用 Azure Stack Hub 版本 2005,除了按照本部分中的步骤操作外,还需要添加面向存储服务 API 版本 2019-02-02 的代码。 若要了解如何面向特定的存储 API 版本,请参阅 GitHub 中的此示例

创建 Azure 存储和 Blob 容器

我们将使用 Azure 存储作为检查点存储。 使用以下步骤创建 Azure 存储帐户。

  1. 创建 Azure 存储帐户

  2. 创建 Blob 容器

  3. 获取存储帐户的连接字符串

    记下连接字符串和容器名称。 你将在接收代码中使用它们。

为接收方创建 Visual Studio 项目

  1. 在“解决方案资源管理器”窗口中,选择并按住 (或右键单击 eventHubQuickStart 解决方案) ,指向“添加”,然后选择“新建项目”。
  2. 选择“ 控制台应用” (.NET“核心) ”,然后选择“ 下一步”。
  3. 输入 PurviewKafkaConsumer“项目名称”,然后选择“ 创建”。

添加事件中心 NuGet 包

  1. 从菜单中选择“ 工具>”“NuGet 包管理器>包管理器控制台 ”。

  2. 运行以下命令以安装 Azure.Messaging.EventHubs NuGet 包:

    Install-Package Azure.Messaging.EventHubs
    
  3. 运行以下命令以安装 Azure.Messaging.EventHubs.Processor NuGet 包:

    Install-Package Azure.Messaging.EventHubs.Processor
    

更新 Main 方法

  1. Program.cs 文件的顶部添加以下using语句。

    using System;
    using System.Text;
    using System.Threading.Tasks;
    using Azure.Storage.Blobs;
    using Azure.Messaging.EventHubs;
    using Azure.Messaging.EventHubs.Consumer;
    using Azure.Messaging.EventHubs.Processor;
    
  2. 将常量添加到Program事件中心连接字符串和事件中心名称的 类。 将括号中的占位符替换为创建事件中心和存储帐户时获得的实际值, (访问密钥 - 主连接字符串) 。 请确保 {Event Hubs namespace connection string} 是命名空间级别的连接字符串,而不是事件中心字符串。

        private const string ehubNamespaceConnectionString = "<EVENT HUBS NAMESPACE - CONNECTION STRING>";
        private const string eventHubName = "<EVENT HUB NAME>";
        private const string blobStorageConnectionString = "<AZURE STORAGE CONNECTION STRING>";
        private const string blobContainerName = "<BLOB CONTAINER NAME>";
    

    消息发送到 Microsoft Purview 时,使用 ATLAS_ENTITIES 作为事件中心名称。

  3. Main 方法替换为以下 async Main 方法。 有关详细信息,请参阅代码中的注释。

        static async Task Main()
        {
            // Read from the default consumer group: $Default
            string consumerGroup = EventHubConsumerClient.DefaultConsumerGroupName;
    
            // Create a blob container client that the event processor will use 
            BlobContainerClient storageClient = new BlobContainerClient(blobStorageConnectionString, blobContainerName);
    
            // Create an event processor client to process events in the event hub
            EventProcessorClient processor = new EventProcessorClient(storageClient, consumerGroup, ehubNamespaceConnectionString, eventHubName);
    
            // Register handlers for processing events and handling errors
            processor.ProcessEventAsync += ProcessEventHandler;
            processor.ProcessErrorAsync += ProcessErrorHandler;
    
            // Start the processing
            await processor.StartProcessingAsync();
    
            // Wait for 10 seconds for the events to be processed
            await Task.Delay(TimeSpan.FromSeconds(10));
    
            // Stop the processing
            await processor.StopProcessingAsync();
        }    
    
  4. 现在,将以下事件和错误处理程序方法添加到 类。

        static async Task ProcessEventHandler(ProcessEventArgs eventArgs)
        {
            // Write the body of the event to the console window
            Console.WriteLine("\tReceived event: {0}", Encoding.UTF8.GetString(eventArgs.Data.Body.ToArray()));
    
            // Update checkpoint in the blob storage so that the app receives only new events the next time it's run
            await eventArgs.UpdateCheckpointAsync(eventArgs.CancellationToken);
        }
    
        static Task ProcessErrorHandler(ProcessErrorEventArgs eventArgs)
        {
            // Write details about the error to the console window
            Console.WriteLine($"\tPartition '{ eventArgs.PartitionId}': an unhandled exception was encountered. This was not expected to happen.");
            Console.WriteLine(eventArgs.Exception.Message);
            return Task.CompletedTask;
        }    
    
  5. 生成项目。 确保没有任何错误。

    备注

    有关包含详细信息注释的完整源代码,请参阅 GitHub 上的此文件

  6. 运行接收器应用程序。

从 Microsoft Purview 接收的消息示例

{
	"version":
		{"version":"1.0.0",
		 "versionParts":[1]
		},
		 "msgCompressionKind":"NONE",
		 "msgSplitIdx":1,
		 "msgSplitCount":1,
		 "msgSourceIP":"10.244.155.5",
		 "msgCreatedBy":
		 "",
		 "msgCreationTime":1618588940869,
		 "message":{
			"type":"ENTITY_NOTIFICATION_V2",
			"entity":{
				"typeName":"azure_sql_table",
					"attributes":{
						"owner":"admin",
						"createTime":0,
						"qualifiedName":"mssql://nayenamakafka.eventhub.sql.net/salespool/dbo/SalesOrderTable",
						"name":"SalesOrderTable",
						"description":"Sales Order Table"
						},
						"guid":"ead5abc7-00a4-4d81-8432-d5f6f6f60000",
						"status":"ACTIVE",
						"displayText":"SalesOrderTable"
					},
					"operationType":"ENTITY_UPDATE",
					"eventTime":1618588940567
				}
}

后续步骤

查看 GitHub 中的更多示例。