C＃でAvroファイルを非直列化する

Question

C＃でApache Avroファイルを逆シリアル化する方法が見つかりません。 Avroファイルは、Microsoft AzureイベントハブのArchive featureによって生成されたファイルです。

私はJSONにファイルを変換するためのApacheから Avro Toolsを使用できるJavaで

：私はSequenceNumber、OffsetとEnqueuedTimeUtcを抽出することができる午前

Microsoft.Hadoop.Avro NuGetパッケージ を使用して

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

が、私からBodyに使用するタイプがわからない場合は例外がスローされます。私はDictionary<string, object>と他のタイプで試しました。

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

スキーマは次のようになります。この、Bodyはnullとbytesの労働組合であるにもかかわらず

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

：

{ 
    "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

私はdynamicを使用して完全なデータアクセスを行うことができました。未加工のbodyデータにアクセスするためのコードは、バイトの配列として格納されています。私の場合には、これらのバイトは、JSONをUTF8でエンコード含まれているが、もちろん、それはあなたが最初にあなたがイベントハブに公開し、あなたの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}"); 
     } 
    } 
} 

誰かが静的に型付けされたソリューションを投稿することができた場合に、私はそれをアップヴォートしますが、どんなシステムでも待ち時間が長いほどイベントハブアーカイブのブロブとの接続が確実になるので、パースのパフォーマンスについては心配しません。 :)

Answer 2

あなたの残りの種類、私は疑うように定義する必要がありますnullablebyte[]にマップします。

C＃では、配列は常に参照型であるため、nullとなり、契約は成立します。

Answer 3

私はこれをApache C＃ライブラリ/フレームワークで動作させることができました。
Azureイベントハブのキャプチャ機能がメッセージコンテンツを含まないファイルを出力することがあるため、しばらく立ち往生しました。 メッセージがもともとEventDataオブジェクトにシリアル化されていた方法に問題があったかもしれません。
以下のコードは、キャプチャブロブコンテナからディスクに保存されたファイルを対象としています。

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イベントハブキャプチャ機能は、すべてのイベントをバックアップするための素晴らしい機能であることに同意します。 Stream Analyticの仕事の出力と同じようにフォーマットをオプションにする必要があると思っていますが、おそらく私はAvroに慣れていきます。

Answer 4

このGistは、.NET Framework 4.5と.NET Standard 1の両方の利点を持つMicrosoft.Hadoop.Avro2を使用してC＃でイベントハブキャプチャをどのように逆シリアル化するかを示しています。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作品）
  • WindowsAzure.Storage（8.3.0作品）

• 名前空間：

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