如何用spark编写avro文件？

0yycz8jy 于 2021-05-29 发布在 Hadoop

关注(0)|答案(1)|浏览(497)

我有一个表示avro模式的数组[byte]。我正试图用spark将它作为avro文件写入hdfs。代码如下：

val values = messages.map(row => (null,AvroUtils.decode(row._2,topic)))
    .saveAsHadoopFile(
      outputPath,
      classOf[org.apache.hadoop.io.NullWritable],
      classOf[CrashPacket],
      classOf[AvroOutputFormat[SpecificRecordBase]]
    )

第2行是数组[字节]
我得到这个错误： org.apache.spark.SparkException: Job aborted due to stage failure: Task 4 in stage 1.0 failed 4 times, most recent failure: Lost task 4.3 in stage 1.0 (TID 98, bdac1nodec06.servizi.gr-u.it): java.lang.NullPointerException at java.io.StringReader.<init>(StringReader.java:50) at org.apache.avro.Schema$Parser.parse(Schema.java:958) at org.apache.avro.Schema.parse(Schema.java:1010) at org.apache.avro.mapred.AvroJob.getOutputSchema(AvroJob.java:143) at org.apache.avro.mapred.AvroOutputFormat.getRecordWriter(AvroOutputFormat.java:153) at org.apache.spark.SparkHadoopWriter.open(SparkHadoopWriter.scala:91) at org.apache.spark.rdd.PairRDDFunctions$$anonfun$13.apply(PairRDDFunctions.scala:1068) at org.apache.spark.rdd.PairRDDFunctions$$anonfun$13.apply(PairRDDFunctions.scala:1059) at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:61) at org.apache.spark.scheduler.Task.run(Task.scala:64) at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:203) at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142) at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617) at java.lang.Thread.run(Thread.java:745)

hadoop avro apache-spark spark-avro

来源：https://stackoverflow.com/questions/39454591/how-to-write-an-avro-file-with-spark

1条答案

按热度按时间

nimxete21#

考虑一下，有一个avro类 StringPair 带构造函数 StringPair(String a, String b) . 然后将记录写入avro文件的代码可以如下所示：

import com.test.{StringPair}
import org.apache.avro.Schema
import org.apache.avro.mapred.{AvroValue, AvroKey}
import org.apache.avro.mapreduce.{AvroKeyValueOutputFormat, AvroJob}
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.hadoop.mapreduce.Job

object TestWriteAvro {
  def main (args: Array[String]){
    val sparkConf = new SparkConf()
    val sc = new SparkContext(sparkConf)
    val job = new Job(sc.hadoopConfiguration)

    AvroJob.setOutputKeySchema(job, Schema.create(Schema.Type.STRING))
    AvroJob.setOutputValueSchema(job, StringPair.getClassSchema)
    val myRdd = sc
      .parallelize(List("1,2", "3,4"))
      .map(x => (x.split(",")(0), x.split(",")(1)))
      .map {case (x, y) => (new AvroKey[String](x), new AvroValue[StringPair](new StringPair(x, y)))}

    myRdd.saveAsNewAPIHadoopFile(args(0), classOf[AvroKey[_]], classOf[AvroValue[_]], classOf[AvroKeyValueOutputFormat[_, _]], job.getConfiguration)

    sc.stop()
  }
}

赞(0）回复(0）举报 2021-05-29

我来回答

如何用spark编写avro文件？

1条答案

相关问题

热门标签

最新问答