のjava.lang.NullPointerException：スパークストリーミング仕事で仕事

Question

ストリーミングスパークに寄木細工のファイルを書き込み中にwriteSupportClassがnullであってはならない、私は以下のコードを使用してHadoopののHDFSで寄木細工のファイルに私のRDDデータを保存しています：

readyToSave.foreachRDD((VoidFunction<JavaPairRDD<Void, MyProtoRecord>>) rdd -> {   
      Configuration configuration = rdd.context().hadoopConfiguration(); 
      Job job = Job.getInstance(configuration); 
      ParquetOutputFormat.setWriteSupportClass(job, ProtoWriteSupport.class); 
      ProtoParquetOutputFormat.setProtobufClass(job, MyProtoRecord.class); 
      rdd.saveAsNewAPIHadoopFile("path-to-hdfs", Void.class, MyProtoRecord.class, ParquetOutputFormat.class, configuration); 
    }); 

と私は以下の例外を取得：

java.lang.NullPointerException: writeSupportClass should not be null 
at parquet.Preconditions.checkNotNull(Preconditions.java:38) 
at parquet.hadoop.ParquetOutputFormat.getWriteSupport(ParquetOutputFormat.java:326) 
at parquet.hadoop.ParquetOutputFormat.getRecordWriter(ParquetOutputFormat.java:272) 
at parquet.hadoop.ParquetOutputFormat.getRecordWriter(ParquetOutputFormat.java:262) 
at org.apache.spark.rdd.PairRDDFunctions$$anonfun$saveAsNewAPIHadoopDataset$1$$anonfun$12.apply(PairRDDFunctions.scala:1112) 
at org.apache.spark.rdd.PairRDDFunctions$$anonfun$saveAsNewAPIHadoopDataset$1$$anonfun$12.apply(PairRDDFunctions.scala:1095) 
at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:70) 
at org.apache.spark.scheduler.Task.run(Task.scala:86) 
at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:274) 
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) 

どのように私はこの問題を解決することができますか？

Answer 1

問題が見つかりました！

readyToSave.foreachRDD((VoidFunction<JavaPairRDD<Void, MyProtoRecord>>) rdd -> { 
      Configuration configuration = rdd.context().hadoopConfiguration(); 
      Job job = Job.getInstance(configuration); 
      ParquetOutputFormat.setWriteSupportClass(job, ProtoWriteSupport.class); 
      ProtoParquetOutputFormat.setProtobufClass(job, MyProtoRecord.class); 
      rdd.saveAsNewAPIHadoopFile("path-to-hdfs", Void.class, MyProtoRecord.class, ParquetOutputFormat.class, job.getConfiguration()); 
    }); 

： " saveAsNewAPIHadoopFile（）メソッドを呼び出している間、あなたの仕事の設定（ job.getConfiguration（））を指定sould