Spark:写入Avro文件

Question

我在Spark,我有一个Avro文件的RDD.我现在想对该RDD进行一些转换并将其保存为Avro文件:

val job = new Job(new Configuration())
AvroJob.setOutputKeySchema(job, getOutputSchema(inputSchema))

rdd.map(elem => (new SparkAvroKey(doTransformation(elem._1)), elem._2))
   .saveAsNewAPIHadoopFile(outputPath, 
  classOf[AvroKey[GenericRecord]], 
  classOf[org.apache.hadoop.io.NullWritable], 
  classOf[AvroKeyOutputFormat[GenericRecord]], 
  job.getConfiguration)

运行时,Spark会抱怨Schema $ recordSchema不可序列化.

如果我取消注释.map调用(并且只有rdd.saveAsNewAPIHadoopFile),则调用成功.

我在这做错了什么？

任何的想法？

Answer 1

此处的问题与作业中使用的avro.Schema类的不可序列化有关.当您尝试从map函数内的代码引用架构对象时,抛出异常.

例如,如果您尝试执行以下操作,您将获得"Task not serializable"异常:

val schema = new Schema.Parser().parse(new File(jsonSchema))
...
rdd.map(t => {
  // reference to the schema object declared outside
  val record = new GenericData.Record(schema)
})

您可以通过在功能块内创建模式的新实例来使一切工作正常:

val schema = new Schema.Parser().parse(new File(jsonSchema))
// The schema above should not be used in closures, it's for other purposes
...
rdd.map(t => {
  // create a new Schema object
  val innserSchema = new Schema.Parser().parse(new File(jsonSchema))
  val record = new GenericData.Record(innserSchema)
  ...
})

由于您不希望为您处理的每条记录解析avro架构,因此更好的解决方案是在分区级别解析架构.以下也有效:

val schema = new Schema.Parser().parse(new File(jsonSchema))
// The schema above should not be used in closures, it's for other purposes
...
rdd.mapPartitions(tuples => {
  // create a new Schema object
  val innserSchema = new Schema.Parser().parse(new File(jsonSchema))

  tuples.map(t => {
    val record = new GenericData.Record(innserSchema)
    ...
    // this closure will be bundled together with the outer one 
    // (no serialization issues)
  })
})

只要您提供对jsonSchema文件的可移植引用,上面的代码就可以工作,因为map函数将由多个远程执行程序执行.它可以是对HDFS中文件的引用,也可以与JAR中的应用程序一起打包(在后一种情况下,您将使用类加载器函数来获取其内容).

对于那些试图将Avro与Spark一起使用的人,请注意仍然存在一些未解决的编译问题,您必须在Maven POM上使用以下导入:

<dependency>
  <groupId>org.apache.avro</groupId>
  <artifactId>avro-mapred</artifactId>
  <version>1.7.7</version>
  <classifier>hadoop2</classifier>
<dependency>

注意"hadoop2"分类器.您可以在https://issues.apache.org/jira/browse/SPARK-3039上跟踪该问题.

Answer 2

Spark 使用的默认序列化器是 Java 序列化。因此，对于所有 java 类型，它将尝试使用 Java 序列化进行序列化。AvroKey 不可序列化，因此您会收到错误。

您可以在自定义序列化中使用 KryoSerializer 或插件（如 Avro）。您可以在此处阅读有关序列化的更多信息。http://spark-project.org/docs/latest/tuning.html

您还可以用可外部化的东西包装您的对象。例如，查看包装 AvroFlumeEvent 的 SparkFlumeEvent： https: //github.com/apache/spark/blob/master/external/flume/src/main/scala/org/apache/spark/streaming/flume/FlumeInputDStream.scala