read - Python Spark avro

spark read format avro

Question

Al intentar escribir avro, me aparece el siguiente error:

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

Había leído en un archivo avro con 3 registros usando:

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()

Luego intenté escribir un solo registro (salida guardada en avro) con:

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)

¿Cómo puedo evitar ese error / escribir un registro avro individual con éxito? Sé que mi esquema es correcto porque es del mismo avro.

Answer 1

Parece que esto no es compatible en este momento. Ahora está intentando usar el mapa de Java como un Registro de Avro y convertirlo en un mapa de Java nuevamente. Es por eso que obtienes el error El error sobre el hashmap java.

Hay una solicitud de extracción de staslos para agregar el formato de salida de Avro, vea el link para la solicitud de extracción y el ejemplo.

Hay un convertidor requerido que falta en AvroConverters.scala para convertir del mapa java al formato avro.

Answer 2

Tuve exactamente el mismo problema la semana pasada. Basado en la respuesta de rfkortekass, usé la solicitud de extracción de staslos e intenté configurar un ejemplo simple de lectura / escritura en archivos avro.

Me las arreglé para obtener una simple demostración de trabajo. Puede encontrar los archivos aquí https://github.com/totor31/spark-avro-python-converters

No sé nada de scala, java ni maven, así que esto es muy experimental: mi objetivo principal era extraer una cantidad muy limitada de archivos del directorio de ejemplos de chispas, para obtener un marco de compilación que permita generar un archivo jar utilizable.

Si alguien con más conocimientos quiere interactuar con mi repositorio, no dude en preguntar.

Muchas gracias a rfkortekass por señalar la solicitud de extracción, me ahorró mucho tiempo.

Answer 3

Volviendo unos meses más tarde:

El uso de los DataFrames de spark en la nueva versión es una buena manera de escribir archivos avro: consulte https://github.com/databricks/spark-avro