Десериализация Avro Spark - PullRequest
Новая
       46

Десериализация Avro Spark

1 голос
/ 07 ноября 2019

Я отправляю поток данных в Azure EventHub со следующим кодом, использующим Microsoft.Hadoop.Avro .. этот код запускается каждые 5 секунд, и просто выводит те же два сериализованных элемента Avro ??: 

  var strSchema = File.ReadAllText("schema.json");
  var avroSerializer = AvroSerializer.CreateGeneric(strSchema);
  var rootSchema = avroSerializer.WriterSchema as RecordSchema;

  var itemList = new List<AvroRecord>();

  dynamic record_one = new AvroRecord(rootSchema);
  record_one.FirstName = "Some";
  record_one.LastName = "Guy";
  itemList.Add(record_one);

  dynamic record_two = new AvroRecord(rootSchema);
  record_two.FirstName = "A.";
  record_two.LastName = "Person";
  itemList.Add(record_two);

  using (var buffer = new MemoryStream())
  {
      using (var writer = AvroContainer.CreateGenericWriter(strSchema, buffer, Codec.Null))
      {
          using (var streamWriter = new SequentialWriter<object>(writer, itemList.Count))
          {
              foreach (var item in itemList)
              {
                  streamWriter.Write(item);
              }
          }
      }

      eventHubClient.SendAsync(new EventData(buffer.ToArray()));
  }

Используемая здесь схема, опять же, v. Simple: 

{
  "type": "record",
  "name": "User",
  "namespace": "SerDes",
  "fields": [
    {
      "name": "FirstName",
      "type": "string"
    },
    {
      "name": "LastName",
      "type": "string"
    }
  ]
}

Я проверил, что это все хорошо, с простым представлением в Azure Stream Analytics на портале:

Stream Analytics Screenshot

Пока все хорошо, но я не могу, поскольку моя жизнь правильно десериализовать это в Databricks использовать команду from_avro() под Scala ..

Загрузить(точно такая же) схема в виде строки: 

val sampleJsonSchema = dbutils.fs.head("/mnt/schemas/schema.json")

Настройка EventHub 

val connectionString = ConnectionStringBuilder("<CONNECTION_STRING>")
  .setEventHubName("<NAME_OF_EVENT_HUB>")
  .build

val eventHubsConf = EventHubsConf(connectionString).setStartingPosition(EventPosition.fromEndOfStream)
val eventhubs = spark.readStream.format("eventhubs").options(eventHubsConf.toMap).load()

Считывание данных .. 

// this works, and i can see the serialised data
display(eventhubs.select($"body"))

// this fails, and with an exception: org.apache.spark.SparkException: Malformed records are detected in record parsing. Current parse Mode: FAILFAST. To process malformed records as null result, try setting the option 'mode' as 'PERMISSIVE'.
display(eventhubs.select(from_avro($"body", sampleJsonSchema)))

По сути, что происходитздесь .. я сериализирую данные с той же схемой, что и десериализация, но что-то искажено .. документация на этом фронте невероятно редка (очень и очень минимальна на веб-сайте Microsoft).

...