本快速入门介绍如何发送和接收 Atlas Kafka 主题事件。 我们将使用 Azure 事件中心和Azure.Messaging.EventHubs .NET 库。
先决条件
如果不熟悉事件中心，请在完成本快速入门之前查看 事件中心概述 。
若要遵循本快速入门，需要满足某些先决条件：
- Azure 订阅。 若要使用 Azure 服务（包括事件中心），需要一个 Azure 订阅。 如果没有 Azure 帐户，可以在创建帐户时注册试用版或使用 MSDN 订阅者权益。
-
Microsoft Visual Studio 2022。 事件中心客户端库利用 C# 8.0 中引入的新功能。 你仍然可以将库与以前的 C# 版本一起使用，但新语法将不可用。 若要使用完整的语法，建议使用 .NET Core SDK 3.0 或更高版本以及语言 版本 设置为
latest进行编译。 如果使用的是 Visual Studio 2019 之前的 Visual Studio 版本，则它没有生成 C# 8.0 项目所需的工具。 Visual Studio 2022（包括免费社区版）可以 在此处下载。 - 活动 Microsoft Purview 帐户。
- 配置了 Microsoft Purview 帐户以发送和接收消息的事件中心：
将消息发布到 Microsoft Purview
让我们创建一个 .NET Core 控制台应用程序，该应用程序通过事件中心 Kafka 主题将事件发送到 Microsoft Purview，ATLAS_HOOK。
若要将消息发布到 Microsoft Purview，需要托管事件中心或至少一个具有挂钩配置的事件中心。
创建 Visual Studio 项目
接下来，在 Visual Studio 中创建 C# .NET 控制台应用程序：
- 启动 Visual Studio。
- 在“开始”窗口中，选择“创建新项目>控制台应用”（.NET Framework）。 需要 .NET 版本 4.5.2 或更高版本。
- 在 项目名称中，输入 PurviewKafkaProducer。
- 选择“ 创建 ”以创建项目。
创建控制台应用程序
- 启动 Visual Studio 2022。
- 选择“ 创建新项目”。
- 在“ 创建新项目 ”对话框中，执行以下步骤：如果未看到此对话框，请在菜单上选择“ 文件 ”，选择“ 新建”，然后选择“ 项目”。
- 为编程语言选择 C# 。
- 选择应用程序的类型的 控制台 。
- 从结果列表中选择 控制台应用（.NET Core ）。
- 然后选择下一步。
添加事件中心 NuGet 包
从菜单中选择“工具>NuGet 包管理器包管理器>控制台”。
运行以下命令以安装 Azure.Messaging.EventHubs NuGet 包和 Azure.Messaging.EventHubs.Producer NuGet 包：
Install-Package Azure.Messaging.EventHubsInstall-Package Azure.Messaging.EventHubs.Producer
编写将消息发送到事件中心的代码
将以下
using语句添加到 Program.cs 文件的顶部：using System; using System.Text; using System.Threading.Tasks; using Azure.Messaging.EventHubs; using Azure.Messaging.EventHubs.Producer;将常量添加到
Program事件中心连接字符串和事件中心名称的类。private const string connectionString = "<EVENT HUBS NAMESPACE - CONNECTION STRING>"; private const string eventHubName = "<EVENT HUB NAME>";将
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."); }生成项目。 确保没有错误。
运行程序并等待确认消息。
注释
有关包含详细信息注释的完整源代码，请参阅 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 客户端库在 Azure（2019-07-07-07 发布时）使用最高可用版本。 如果使用 Azure Stack Hub 版本 2005，除了按照本节中的步骤作外，还需要添加面向存储服务 API 版本 2019-02-02 的代码。 若要了解如何以特定存储 API 版本为目标，请参阅 GitHub 中的此示例。
创建 Azure 存储和 Blob 容器
我们将使用 Azure 存储作为检查点存储。 使用以下步骤创建 Azure 存储帐户。
-
记下连接字符串和容器名称。 你将在接收代码中使用它们。
为接收方创建 Visual Studio 项目
- 在“解决方案资源管理器”窗口中，选择并按住 EventHubQuickStart 解决方案，指向 “添加”，然后选择“ 新建项目”。
- 选择 控制台应用（.NET Core），然后选择“ 下一步”。
- 输入 PurviewKafkaConsumer 以获取 项目名称，然后选择“ 创建”。
添加事件中心 NuGet 包
从菜单中选择“工具>NuGet 包管理器包管理器>控制台”。
运行以下命令以安装 Azure.Messaging.EventHubs NuGet 包：
Install-Package Azure.Messaging.EventHubs运行以下命令以安装 Azure.Messaging.EventHubs.Processor NuGet 包：
Install-Package Azure.Messaging.EventHubs.Processor
更新 Main 方法
在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;将常量添加到
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 作为事件中心名称。
将
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(); }现在，将以下事件和错误处理程序方法添加到类。
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; }生成项目。 确保没有错误。
注释
有关包含详细信息注释的完整源代码，请参阅 GitHub 上的此文件。
运行接收方应用程序。
从 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 中的更多示例。