Microsoft.Azure.Data.SchemaRegistry.ApacheAvro 1.0.1

Azure Schema Registry Apache Avro .NET客户端库

Azure Schema Registry是一个由Azure Event Hubs托管的模式存储库服务，提供模式存储、版本控制和管理工作。此包提供了一种能够序列化和反序列化包含Schema Registry方案标识符和Avro序列化数据的有效负载的Avro序列化器。

入门

安装包

使用NuGet安装.NET的Azure Schema Registry Apache Avro库

dotnet add package Microsoft.Azure.Data.SchemaRegistry.ApacheAvro

先决条件

• Azure订阅
• Event Hubs命名空间

如果您需要创建名称空间，可以使用 Azure 门户或 Azure 事件中心名称空间的创建 或 Azure PowerShell。

您可以使用以下命令使用 Azure PowerShell 创建事件中心名称空间：

New-AzEventHubNamespace -ResourceGroupName myResourceGroup -NamespaceName namespace_name -Location eastus

身份验证客户端

为了与 Azure 模式注册表服务交互，您需要创建 模式注册表客户端类 的一个实例。要创建此客户端，您需要 Azure 资源凭据和事件中心名称空间的宿主机名。

获取凭据

有关获取已验证凭据并开始与 Azure 资源交互的说明，请参阅此处快速入门指南 快速入门向导。

获取事件中心名称空间宿主机名

最简单的方法是使用 Azure 门户 并导航到您的事件中心名称空间。从“概览”选项卡，您会看到 宿主机名。从该字段复制值。

创建 SchemaRegistryClient

一旦您拥有 Azure 资源凭据和事件中心名称空间的宿主机名，就可以创建 SchemaRegistryClient。您还需要 Azure.Identity 包以创建凭据。

// Create a new SchemaRegistry client using the default credential from Azure.Identity using environment variables previously set,
// including AZURE_CLIENT_ID, AZURE_CLIENT_SECRET, and AZURE_TENANT_ID.
// For more information on Azure.Identity usage, see: https://github.com/Azure/azure-sdk-for-net/blob/main/sdk/identity/Azure.Identity/README.md
var schemaRegistryClient = new SchemaRegistryClient(fullyQualifiedNamespace: fullyQualifiedNamespace, credential: new DefaultAzureCredential());

关键概念

序列化器

此库提供了一个序列化器，SchemaRegistryAvroSerializer，该序列化器与 EventData 事件交互。SchemaRegistryAvroSerializer 使用 SchemaRegistryClient 来丰富使用该模式序列化数据的模式 ID。

此序列化器需要 Apache Avro 库。此序列化器接受的负载类型包括 GenericRecord 和 ISpecificRecord。

示例

以下显示了通过 SchemaRegistryAvroSerializer 可用的示例。这些操作既提供同步又提供异步方法。这些示例使用使用此方案生成的 Apache Avro 类 Employee.cs。

{
   "type" : "record",
    "namespace" : "TestSchema",
    "name" : "Employee",
    "fields" : [
        { "name" : "Name" , "type" : "string" },
        { "name" : "Age", "type" : "int" }
    ]
}

有关使用 Apache Avro 库生成类的详细信息，请参阅 Avro C# 文档。

使用事件中心 EventData 模型序列化和反序列化数据

要使用 Avro 信息序列化 EventData 实例，可以执行以下操作

var serializer = new SchemaRegistryAvroSerializer(client, groupName, new SchemaRegistryAvroSerializerOptions { AutoRegisterSchemas = true });

var employee = new Employee { Age = 42, Name = "Caketown" };
EventData eventData = (EventData) await serializer.SerializeAsync(employee, messageType: typeof(EventData));

// the schema Id will be included as a parameter of the content type
Console.WriteLine(eventData.ContentType);

// the serialized Avro data will be stored in the EventBody
Console.WriteLine(eventData.EventBody);

// construct a publisher and publish the events to our event hub
var fullyQualifiedNamespace = "<< FULLY-QUALIFIED EVENT HUBS NAMESPACE (like something.servicebus.windows.net) >>";
var eventHubName = "<< NAME OF THE EVENT HUB >>";
var credential = new DefaultAzureCredential();

// It is recommended that you cache the Event Hubs clients for the lifetime of your
// application, closing or disposing when application ends.  This example disposes
// after the immediate scope for simplicity.

await using var producer = new EventHubProducerClient(fullyQualifiedNamespace, eventHubName, credential);
await producer.SendAsync(new EventData[] { eventData });

要反序列化您正在消费的事件数据 EventData 事件

// construct a consumer and consume the event from our event hub

// It is recommended that you cache the Event Hubs clients for the lifetime of your
// application, closing or disposing when application ends.  This example disposes
// after the immediate scope for simplicity.

await using var consumer = new EventHubConsumerClient(EventHubConsumerClient.DefaultConsumerGroupName, fullyQualifiedNamespace, eventHubName, credential);
await foreach (PartitionEvent receivedEvent in consumer.ReadEventsAsync())
{
    Employee deserialized = (Employee) await serializer.DeserializeAsync(eventData, typeof(Employee));
    Console.WriteLine(deserialized.Age);
    Console.WriteLine(deserialized.Name);
    break;
}

您还可以使用通用方法来序列化和反序列化数据。如果您不是在 Avro 序列化器之上构建库，这可能更方便，因为您不必担心泛型的传染性

var serializer = new SchemaRegistryAvroSerializer(client, groupName, new SchemaRegistryAvroSerializerOptions { AutoRegisterSchemas = true });

var employee = new Employee { Age = 42, Name = "Caketown" };
EventData eventData = await serializer.SerializeAsync<EventData, Employee>(employee);

// the schema Id will be included as a parameter of the content type
Console.WriteLine(eventData.ContentType);

// the serialized Avro data will be stored in the EventBody
Console.WriteLine(eventData.EventBody);

类似地，进行反序列化

Employee deserialized = await serializer.DeserializeAsync<Employee>(eventData);
Console.WriteLine(deserialized.Age);
Console.WriteLine(deserialized.Name);

直接使用 MessageContent 序列化和反序列化数据

您还可以使用 MessageContent 进行序列化和反序列化。如果未集成任何与 MessageContent 一起工作的消息库，请使用此选项。

var serializer = new SchemaRegistryAvroSerializer(client, groupName, new SchemaRegistryAvroSerializerOptions { AutoRegisterSchemas = true });
MessageContent content = await serializer.SerializeAsync<MessageContent, Employee>(employee);

Employee deserializedEmployee = await serializer.DeserializeAsync<Employee>(content);

故障排除

如果在与模式注册表服务通信时遇到错误，这些错误将以RequestFailedException的形式抛出。序列化器仅在第一次遇到模式（在序列化时）或模式ID（在反序列化时）时才会与该服务通信。与无效内容类型相关的任何错误将以FormatException的形式抛出。与无效模式相关的错误将以Exception的形式抛出，并且InnerException属性将包含由Apache Avro库抛出的基本异常。此类错误通常在测试期间捕获，不应在代码中处理。与不兼容模式相关的任何错误将以包含设置有Apache Avro库基本异常的InnerException属性的Exception的形式抛出。

下一步

有关更多信息，请参阅Azure Schema Registry。

贡献

此项目欢迎贡献和建议。大多数贡献要求您同意一项贡献者许可协议（CLA），声明您有权并且实际上已经向我们授予使用您的贡献的权利。有关详细信息，请访问cla.microsoft.com。

当您提交请求时，CLA机器人会自动判断您是否需要提供CLA，并相应地标记PR（例如，标签、注释）。只需遵循机器人提供的说明。您在整个使用我们的CLA的仓库中只需这样做一次。

此项目采用了Microsoft Open Source Code of Conduct。有关更多信息，请参阅行为准则FAQ或通过[email protected]联系以获取任何附加问题或评论。