Deserialize một tập tin Avro với C#

Question

Tôi không thể tìm thấy một cách để deserialize một tập tin Avro Apache với C#. Tệp Avro là tệp được tạo bởi Archive feature trong Trung tâm sự kiện Microsoft Azure.

Với Java tôi có thể sử dụng Avro Tools từ Apache để chuyển đổi các tập tin để JSON:

java -jar avro-tools-1.8.1.jar tojson --pretty inputfile > output.json 

Sử dụng NuGet gói Microsoft.Hadoop.Avro ​​ tôi có thể trích xuất SequenceNumber, Offset và EnqueuedTimeUtc, nhưng kể từ khi tôi không biết sử dụng loại nào cho Body một ngoại lệ được ném. Tôi đã thử với Dictionary<string, object> và các loại khác.

static void Main(string[] args) 
{ 
    var fileName = "..."; 

    using (Stream stream = new FileStream(fileName, FileMode.Open, FileAccess.Read, FileShare.Read)) 
    { 
     using (var reader = AvroContainer.CreateReader<EventData>(stream)) 
     { 
      using (var streamReader = new SequentialReader<EventData>(reader)) 
      { 
       var record = streamReader.Objects.FirstOrDefault(); 
      } 
     } 
    } 
} 

[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")] 
    public foo Body { get; set; } 

    // More properties... 
} 

Giản đồ trông như thế này:

{ 
    "type": "record", 
    "name": "EventData", 
    "namespace": "Microsoft.ServiceBus.Messaging", 
    "fields": [ 
    { 
     "name": "SequenceNumber", 
     "type": "long" 
    }, 
    { 
     "name": "Offset", 
     "type": "string" 
    }, 
    { 
     "name": "EnqueuedTimeUtc", 
     "type": "string" 
    }, 
    { 
     "name": "SystemProperties", 
     "type": { 
     "type": "map", 
     "values": [ "long", "double", "string", "bytes" ] 
     } 
    }, 
    { 
     "name": "Properties", 
     "type": { 
     "type": "map", 
     "values": [ "long", "double", "string", "bytes" ] 
     } 
    }, 
    { 
     "name": "Body", 
     "type": [ "null", "bytes" ] 
    } 
    ] 
}  

Answer 1

Tôi có thể truy cập dữ liệu đầy đủ khi đang sử dụng dynamic. Đây là mã để truy cập dữ liệu thô body, được lưu trữ dưới dạng một mảng byte. Trong trường hợp của tôi, những byte chứa UTF8 mã hóa JSON, nhưng tất nhiên nó phụ thuộc vào cách bạn tạo ra ban đầu EventData trường hợp của bạn mà bạn xuất bản cho tổ chức sự kiện Hub:

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}"); 
     } 
    } 
} 

Nếu ai đó có thể gửi một giải pháp tĩnh đánh máy, Tôi sẽ upvote nó, nhưng cho rằng độ trễ lớn hơn trong bất kỳ hệ thống gần như chắc chắn sẽ được kết nối với các đốm màu lưu trữ Hub sự kiện, tôi sẽ không lo lắng về phân tích hiệu suất. :)

Answer 2

loại còn lại của bạn, tôi nghi ngờ nên được định nghĩa là:

[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; } 
} 

Mặc dù Body là một liên minh của null và bytes, đây ánh xạ tới một số nullablebyte[].

Trong C#, mảng luôn là loại tham chiếu để có thể là null và hợp đồng đã hoàn thành.

Answer 3

Tôi cuối cùng cũng có thể làm việc này với thư viện/khung công tác Apache C#.
Tôi bị kẹt trong một thời gian vì tính năng Chụp của Trung tâm sự kiện Azure đôi khi xuất ra một tệp mà không có bất kỳ nội dung thư nào. Tôi cũng có thể gặp vấn đề với cách các tin nhắn ban đầu được tuần tự hóa vào đối tượng EventData.
Mã bên dưới dành cho tệp được lưu vào đĩa từ vùng chứa blob chụp.

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

Điều này cũng hoạt động khi sử dụng đối tượng GenericRecord.

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

Việc này cần một số nỗ lực để tìm hiểu. Tuy nhiên, bây giờ tôi đồng ý tính năng này của Azure Event Hubs Capture là một tính năng tuyệt vời để sao lưu tất cả các sự kiện. Tôi vẫn cảm thấy họ nên làm cho các định dạng tùy chọn như họ đã làm với đầu ra công việc phân tích Stream nhưng có lẽ tôi sẽ làm quen với Avro.

Answer 4

Điều này Gist cho biết cách để deserialize chụp trung tâm sự kiện với C# bằng Microsoft.Hadoop.Avro2, có lợi thế là cả .NET Framework 4.5 và .NET Standard 1.6 compliant:

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; } 
} 

tài liệu tham khảo

• NuGet:

  • Microsoft.Hadoop.Avro2 (1.2.1 công trình)
  • WindowsAzure.Storage (8.3.0 công trình)

• Không gian tên:

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