Come leggere il file Avro in PySpark

Question

Sto scrivendo un lavoro spark usando python. Tuttavia, ho bisogno di leggere un sacco di file avro.

This è la soluzione più vicina che ho trovato nella cartella di esempio di Spark. Tuttavia, devi inviare questo script python usando spark-submit. Nella riga di comando di spark-submit, è possibile specificare la classe driver, in tal caso, verrà individuata tutta la classe avrokey, avrovalue.

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

Nel mio caso, ho bisogno di eseguire tutto all'interno dello script Python, ho cercato di creare una variabile di ambiente per includere il file jar, croce dito Python aggiungerà il vaso al percorso, ma chiaramente non lo è , mi sta dando un errore di classe inaspettato.

os.environ['SPARK_SUBMIT_CLASSPATH'] = "/opt/cloudera/parcels/CDH-5.1.0-1.cdh5.1.0.p0.53/lib/spark/examples/lib/spark-examples_2.10-1.0.0-cdh5.1.0.jar" 

Qualcuno può aiutarmi a leggere il file avro in uno script python?

Answer 1

È possibile utilizzare la libreria spark-avro. In primo luogo permette di creare un esempio di dati:

import avro.schema 
from avro.datafile import DataFileReader, DataFileWriter 

schema_string ='''{"namespace": "example.avro", 
"type": "record", 
"name": "KeyValue", 
"fields": [ 
    {"name": "key", "type": "string"}, 
    {"name": "value", "type": ["int", "null"]} 
] 
}''' 

schema = avro.schema.parse(schema_string) 

with open("kv.avro", "w") as f, DataFileWriter(f, DatumWriter(), schema) as wrt: 
    wrt.append({"key": "foo", "value": -1}) 
    wrt.append({"key": "bar", "value": 1}) 

lettura utilizzando spark-csv è semplice come questo:

df = sqlContext.read.format("com.databricks.spark.avro").load("kv.avro") 
df.show() 

## +---+-----+ 
## |key|value| 
## +---+-----+ 
## |foo| -1| 
## |bar| 1| 
## +---+-----+ 

Answer 2

La prima soluzione richiede di installare una terza parte la dipendenza Java, che non è qualcosa di più di Python gli sviluppatori sono felici con. Ma non hai davvero bisogno di una libreria esterna se tutto ciò che vuoi è analizzare i tuoi file Avro con un dato schema. Puoi solo leggere i file binari e analizzarli con il tuo pacchetto Avro Python preferito.

Per esempio, questo è il modo è possibile caricare i file usando Avro fastavro:

from io import BytesIO 
import fastavro 

schema = { 
    ... 
} 

rdd = sc.binaryFiles("/path/to/dataset/*.avro")\ 
    .flatMap(lambda args: fastavro.reader(BytesIO(args[1]), reader_schema=schema)) 

print(rdd.collect())