私はavroスキーマを表す配列[Byte]を持っています。私はそれをsparkでavroファイルとしてHdfsに書き込もうとしています。これはコードです: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)
Sparkでavroファイルを書き込むには?
0
A
答えて
1
はコンストラクタStringPair(String a, String b)
とアブロクラスStringPair
があることを、考えてみましょう:
val values = messages.map(row => (null,AvroUtils.decode(row._2,topic)))
.saveAsHadoopFile(
outputPath,
classOf[org.apache.hadoop.io.NullWritable],
classOf[CrashPacket],
classOf[AvroOutputFormat[SpecificRecordBase]]
)
row._2は、Array [バイト]
私はこのエラーを取得しています。次に、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()
}
}
関連する問題
- 1. Apache Sparkでファイルに書き込む
- 2. Sparkワーカープロセス内にファイルを書き込む
- 3. newAPIHadoopFileを使ってsparkでavroファイルを読むには?
- 4. AVROデータをHadoopのhdfsに書き込む
- 5. データフレームをsparkでavroに書き込む際にエラーが発生しました:NoSuchElementException
- 6. Pyspark Kernelを使ってJupyterノートブックでSpark Avroファイルを読む
- 7. jsonファイルにsparkデータフレームを書き込むことができません。
- 8. Avroへの書き込み時にSparkがスキーマを変更する
- 9. ディレクトリにファイルを書き込むには
- 10. ファイルに書き込むpython
- 11. cURL - ファイルに書き込む
- 12. ファイルに書き込む(Python)
- 13. Spark:Avroファイルに書き込む
- 14. ファイルに書き込む
- 15. ファイルに書き込むキーストローク
- 16. ファイルに書き込む(Python)
- 17. ファイルに書き込む
- 18. Avro/Parquetを使用してHDFSにリアルタイムデータを書き込む方法は?
- 19. マルチスレッドアプリケーションでファイルにログを書き込む
- 20. ストリームライターでファイルにラインを書き込む
- 21. Javaでhdfsにファイルを書き込む
- 22. オブジェクトをC++でファイルに書き込む
- 23. Matlab:aviファイルを書き込む
- 24. Jsonファイルを書き込む
- 25. OCamlファイルに書き込む方法は?
- 26. C#でCSVファイルに書き込む
- 27. MPIでファイルに書き込む
- 28. PythonでUTF-8ファイルに書き込む
- 29. HadoopでHDFSのファイルに書き込む
- 30. Clojureでファイルに書き込む