用c#反序列化Avro文件

我能够使用dynamic获得完整的数据访问工作。下面是访问原始body数据的代码，该数据存储为字节数组。在我的情况下，这些字节包含utf8编码的JSON，但当然这取决于您最初如何创建您发布到事件中心的EventData实例:

using (var reader = AvroContainer.CreateGenericReader(stream))
{
    while (reader.MoveNext())
    {
        foreach (dynamic record in reader.Current.Objects)
        {
            var sequenceNumber = record.SequenceNumber;
            var bodyText = Encoding.UTF8.GetString(record.Body);
            Console.WriteLine($"{sequenceNumber}: {bodyText}");
        }
    }
}

如果有人能发布一个静态类型的解决方案，我会支持它，但考虑到在任何系统中更大的延迟几乎肯定是连接到Event Hub Archive blobs，我不会担心解析性能。:)

这个Gist展示了如何使用Microsoft.Hadoop用c#反序列化事件中心捕获。Avro2，它的优点是同时兼容。net Framework 4.5和。net Standard 1.6:

 var connectionString = "<Azure event hub capture storage account connection string>";
 var containerName = "<Azure event hub capture container name>";
 var blobName = "<Azure event hub capture BLOB name (ends in .avro)>";
 var storageAccount = CloudStorageAccount.Parse(connectionString);
 var blobClient = storageAccount.CreateCloudBlobClient();
 var container = blobClient.GetContainerReference(containerName);
 var blob = container.GetBlockBlobReference(blobName);
 using (var stream = blob.OpenRead())
 using (var reader = AvroContainer.CreateGenericReader(stream))
     while (reader.MoveNext())
         foreach (dynamic result in reader.Current.Objects)
         {
             var record = new AvroEventData(result);
             record.Dump();
         }
 public struct AvroEventData
 {
     public AvroEventData(dynamic record)
     {
         SequenceNumber = (long) record.SequenceNumber;
         Offset = (string) record.Offset;
         DateTime.TryParse((string) record.EnqueuedTimeUtc, out var enqueuedTimeUtc);
         EnqueuedTimeUtc = enqueuedTimeUtc;
         SystemProperties = (Dictionary<string, object>) record.SystemProperties;
         Properties = (Dictionary<string, object>) record.Properties;
         Body = (byte[]) record.Body;
     }
     public long SequenceNumber { get; set; }
     public string Offset { get; set; }
     public DateTime EnqueuedTimeUtc { get; set; }
     public Dictionary<string, object> SystemProperties { get; set; }
     public Dictionary<string, object> Properties { get; set; }
     public byte[] Body { get; set; }
 }

• NuGet引用:

  • Microsoft.Hadoop。Avro2 (1.2.1 works)
  • WindowsAzure。存储(8.3.0可用)

• 名称空间:

  • Microsoft.Hadoop.Avro.Container
  • Microsoft.WindowsAzure.Storage

我终于能够让它与Apache c#库/框架一起工作了。
我被困了一段时间，因为Azure Event Hubs的Capture功能有时会输出一个没有任何消息内容的文件。我可能也有一个问题，如何消息最初序列化到EventData对象。
下面的代码是从捕获blob容器保存到磁盘的文件。

var dataFileReader = DataFileReader<EventData>.OpenReader(file);
foreach (var record in dataFileReader.NextEntries)
{
   // Do work on EventData object
}

使用GenericRecord对象也可以。

var dataFileReader = DataFileReader<GenericRecord>.OpenReader(file);

这需要一些努力才能弄清楚。不过，我现在同意Azure事件中心捕获功能是备份所有事件的一个很好的功能。我仍然觉得他们应该让格式可选，就像他们对流分析作业输出所做的那样，但也许我会习惯Avro。

我建议您使用https://github.com/AdrianStrugala/AvroConvert

和简单:

byte[] avroFileContent = File.ReadAllBytes(fileName);
var result = AvroConvert.Deserialize<EventData>(avroFileContent);

库本身是一个使用Avro格式的开发流程的改进。您甚至不需要模型上的模式或属性。(我是这个库的贡献者)

我想剩下的类型应该定义为:

[DataContract(Namespace = "Microsoft.ServiceBus.Messaging")]
[KnownType(typeof(Dictionary<string, object>))]
public class EventData
{
    [DataMember]
    public IDictionary<string, object> SystemProperties { get; set; }
    [DataMember]
    public IDictionary<string, object> Properties { get; set; }
    [DataMember]
    public byte[] Body { get; set; }
}

尽管Body是null和bytes的并集，但它映射到nullable和byte[]。

在c#中，数组始终是引用类型，因此可以是null，并完成契约。

您还可以使用NullableSchema属性将Body标记为字节和null的联合。这将允许您使用强类型接口。

[DataContract(Namespace = "Microsoft.ServiceBus.Messaging")]
public class EventData
{
    [DataMember(Name = "SequenceNumber")]
    public long SequenceNumber { get; set; }
    [DataMember(Name = "Offset")]
    public string Offset { get; set; }
    [DataMember(Name = "EnqueuedTimeUtc")]
    public string EnqueuedTimeUtc { get; set; }
    [DataMember(Name = "Body")]
    [NullableSchema]
    public foo Body { get; set; }
}