Python Spark avro

При попытке записать avro я получаю следующую ошибку:

org.apache.spark.SparkException: Job aborted due to stage failure: Task 7 in stage 35.0 failed 1 times, most recent failure: Lost task 7.0 in stage 35.0 (TID 110, localhost): java.lang.ClassCastException: java.util.HashMap cannot be cast to org.apache.avro.mapred.AvroWrapper

Я прочитал в файле avro с тремя записями, используя:

avro_rdd = sc.newAPIHadoopFile(
    "threerecords.avro",
    "org.apache.avro.mapreduce.AvroKeyInputFormat",
    "org.apache.avro.mapred.AvroKey",
    "org.apache.hadoop.io.NullWritable",
    keyConverter="org.apache.spark.examples.pythonconverters.AvroWrapperToJavaConverter",
    conf=None)

output = avro_rdd.map(lambda x: x[0]).collect()

Затем я попытался записать одну запись (вывод, хранящийся в avro) с помощью:

conf = {"avro.schema.input.key": reduce(lambda x, y: x + y, sc.textFile("myschema.avsc", 1).collect())}

sc.parallelize([output[0]]).map(lambda x: (x, None)).saveAsNewAPIHadoopFile(
    "output.avro",
    "org.apache.avro.mapreduce.AvroKeyOutputFormat",
    "org.apache.avro.mapred.AvroKey",
    "org.apache.hadoop.io.NullWritable",
    keyConverter="org.apache.spark.examples.pythonconverters.AvroWrapperToJavaConverter",
    conf=conf)

Как мне обойти эту ошибку/выписать индивидуальную запись об авторе? Я знаю, что моя схема правильная, потому что она принадлежит самому автору.

Ответ 1

Похоже, на данный момент это не поддерживается. Теперь вы пытаетесь использовать карту java в качестве записи Avro и снова скрывать ее на карте Java. Вот почему вы получаете сообщение об ошибке в jash hashmap.

Существует запрос на растяжение от staslos, чтобы добавить выходной формат Avro, см. ссылка для запроса на перенос и пример.

Требуется конвертер, который отсутствует в AvroConverters.scala для преобразования из java-карты обратно в формат avro.

Ответ 2

У меня была такая же проблема на прошлой неделе. Основываясь на ответе rfkortekass, я использовал запрос на растяжение из staslos и попытался установить простой пример чтения/записи в файлах avro.

Мне удалось получить простую демонстрационную работу. Вы можете найти файлы здесь https://github.com/totor31/spark-avro-python-converters

Я ничего не знаю для scala, java и maven, поэтому это очень экспериментально: моей главной целью было извлечь очень ограниченное количество файлов из каталога примеров искры, чтобы получить структуру компиляции, которая позволяет генерировать полезную jar файл.

Если кто-то более осведомленный хочет взаимодействовать с моим репозиторием, не стесняйтесь спрашивать.

Огромное спасибо rfkortekass за то, что он указал на запрос на тяну, он мне очень помог.

Ответ 3

Возвращение назад через месяц:

с использованием искры. DataFrames в новой версии - хороший способ писать файлы avro: см. https://github.com/databricks/spark-avro