Как читать закодированные данные Avro Binary (Base64) в Spark Scala - PullRequest
Новая
       29

Как читать закодированные данные Avro Binary (Base64) в Spark Scala

0 голосов
/ 14 октября 2019

Я пытаюсь прочитать файл avro, который закодирован в двоичном формате (Base64), и сжатый Hadoop cat в файле avro выглядит следующим образом: 

Objavro.schema? 
{"type":"record","name":"ConnectDefault","namespace":"xyz.connect.avro","fields": 
[{"name":"service","type":"string"},{"name":"timestamp","type":"long"}, 
{"name":"count","type":"int"},{"name":"encoderKey","type":{"type":"map","values":"string"}}, 
{"name":"schema","type":"string"},{"name":"data","type":"string"}]}>??n]

Мне нужно извлечь и прочитать "схему" и«данные» из вышеуказанного файла. «Схема» связана с «данными», которые имеют несколько полей

. Я попытался выполнить следующие шаги:

1.Чтение двоичного файла 

val binaryFilesRDD = sc.binaryFiles("file+0+00724+00731.avro").map { x => ( x._2.toArray) }
binaryFilesRDD: org.apache.spark.rdd.RDD[Array[Byte]] = MapPartitionsRDD[1] at map at 
<console>:24
Преобразование СДР [Массив [Байт]] в Массив [Байт] 
scala> val newArray = binaryFilesRDD.collect().flatten 
    newArray: Array[Byte] = Array(17, 18, 16, 51, 24, 22, 17, 18, 117, 151, 76, 105, 95, 124....
Вызов следующего метода с использованием newArray (т.е. Array [Byte]) для получения записей из байтов 
    def getGenericRecordfromByte(inputData:Array[Byte], inputDataSchema: Schema): GenericRecord = 
    {
        val datareader = new GenericDatumReader[GenericRecord](inputDataSchema)
        val datadecoder = DecoderFactory.get.binaryDecoder(inputData, null)
        datareader.read(null, datadecoder) 
    }

Но я получаю следующие ошибки. 

    scala> val newDataRecords = getGenericRecordfromByte(newArray,inputDataSchema)
    org.apache.avro.AvroRuntimeException: Malformed data. Length is negative: -40
  at org.apache.avro.io.BinaryDecoder.doReadBytes(BinaryDecoder.java:336)
  at org.apache.avro.io.BinaryDecoder.readString(BinaryDecoder.java:263)
  at org.apache.avro.io.ResolvingDecoder.readString(ResolvingDecoder.java:201)
  at org.apache.avro.generic.GenericDatumReader.readString(GenericDatumReader.java:363)
  at org.apache.avro.generic.GenericDatumReader.readString(GenericDatumReader.java:355)
  at org.apache.avro.generic.GenericDatumReader.read(GenericDatumReader.java:157)
  at org.apache.avro.generic.GenericDatumReader.readField(GenericDatumReader.java:193)
  at org.apache.avro.generic.GenericDatumReader.readRecord(GenericDatumReader.java:183)
  at org.apache.avro.generic.GenericDatumReader.read(GenericDatumReader.java:151)
  at org.apache.avro.generic.GenericDatumReader.read(GenericDatumReader.java:142)

Пожалуйстасовет

1 Ответ

0 голосов
/ 14 октября 2019

Вы запускаете искровую оболочку следующим образом: 

spark-shell --packages org.apache.spark:spark-avro_2.11:2.4.4

Или, может быть, так: 

spark2-shell --packages org.apache.spark:spark-avro_2.11:2.4.4

А затем вы делаете: 

spark.read.format("com.databricks.spark.avro").load("/file/path")
...