如何在 PySpark 中读取 Avro 文件

我正在使用 python 编写 Spark 作业。但是，我需要读取一大堆 avro 文件。

This https://github.com/apache/spark/blob/master/examples/src/main/python/avro_inputformat.py是我在 Spark 示例文件夹中找到的最接近的解决方案。但是，您需要使用spark-submit 提交此python 脚本。在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文件，手指交叉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"

谁能帮我如何在一个 python 脚本中读取 avro 文件？

火花 >= 2.4.0

您可以使用内置 Avro 支持 https://spark.apache.org/docs/latest/sql-data-sources-avro.html。该 API 向后兼容spark-avro包，添加了一些内容（最值得注意的是from_avro / to_avro功能）。

请注意，该模块未与标准 Spark 二进制文件捆绑在一起，必须使用spark.jars.packages或等效机制。

也可以看看Pyspark 2.4.0，使用读取流从 kafka 读取 avro - Python https://stackoverflow.com/q/54693110/10465355

火花

您可以使用spark-avro https://github.com/databricks/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|
## +---+-----+