python Spark avro

Question

尝试编写 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

我使用以下方法读取了一个包含 3 条记录的 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)

如何解决该错误/成功写出单个 avro 记录？我知道我的架构是正确的，因为它来自 avro 本身。

Answer 1

目前似乎不支持此功能。您现在正尝试将 java 映射用作 Avro 记录并再次将其转换为 Java 映射。这就是为什么您会收到有关 java hashmap 错误的错误。

staslos 有一个拉取请求来添加 Avro 输出格式，请参阅 link用于拉取请求和示例。

AvroConverters.scala 中缺少一个转换器，用于从 java 映射转换回 avro 格式。