Как читать файл Avro в PySpark

Я пишу искровую работу с использованием Python. Однако мне нужно прочитать целую кучу файлов avro.

Это самое близкое решение, которое я нашел в папке примеров Spark. Тем не менее, вам нужно отправить этот скрипт на python, используя spark-submit. В командной строке spark-submit вы можете указать драйвер-класс, в этом случае будет расположен весь ваш класс 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)

В моем случае мне нужно запустить все в скрипте Python, я попытался создать переменную окружения, включающую файл jar, метод Cross Cross Python добавит jar к пути, но, очевидно, это не так, он дает мне неожиданный класс ошибка.

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"

Может кто-нибудь помочь мне, как прочитать файл avro в одном скрипте Python?

Ответ 1

Искра> = 2.4.0

Вы можете использовать встроенную поддержку Avro. API обратно совместим с пакетом spark-avro, с несколькими дополнениями (особенно to_avro функции from_avro/to_avro).

Обратите внимание, что модуль не связан со стандартными двоичными файлами Spark и должен быть включен с использованием spark.jars.packages или эквивалентного механизма.

Искра <2.4.0

Вы можете использовать библиотеку spark-avro. Сначала давайте создадим пример набора данных:

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

Читать его с помощью spark-csv так же просто, как это:

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

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

Ответ 2

В предыдущем решении требуется установить стороннюю зависимость Java, которая не является чем-то большим, чем у большинства разработчиков Python. Но вам не нужна внешняя библиотека, если все, что вы хотите сделать, - это проанализировать ваши файлы Avro с помощью данной схемы. Вы можете просто прочитать двоичные файлы и проанализировать их с помощью своего любимого пакета Avro python.

Например, вы можете загружать файлы 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())