作为Spark作业提交时Spark RDD映射中的NullPointerException

Question

18:02:55,271错误UTILS:91-中止任务java.lang.nullpointerException在org.apache.spark.sql.catalyst.expressions.generatedClass$GeneratedIterator.agg_doAggregateWithKeys$（未知源）在org.apache.spark.sql.catalyst.expressions.generatedClass$GeneratedIterator.processNext（未知源）在degenexec.scala:370)在scala.collection.iterator$$Anon$12。在org.apache.spark.sql.execution.datasources.defaultWriterContainer$$AnonFun$WriterOWS$1。在er.Scala:252)在org.apache.spark.util.utils$.trywithSafeFinallyandFailureCallbacks(utils.scala:1325)在org.apache.spark.sql.execution.datasources.defaultwritercontainer.writercontainers:258)在org.apache.spark.sql.execution.datasources.insertintohadoopfsrelationcommand$$anonfun$run$1$$anonfun$apply$mcv$sp$1.在org.apache.spark.sql.execution.datasources SP$1.Apply(InsertIntohadoopfsRelationCommand.scala:143)在org.apache.spark.scheduler.resulttask.runtask(resulttask.scala:70)在org.apache.spark.scheduler.task.run(task.scala:85)在org.apache.spark.executor.executor$taskrunner.run(executor.scala:274)在java.util.concurrent.threadpoolexecutor.runworker.run(thread.java:745)

据我们所知，这是在以下方法中发生的：

def process(dataFrame: DataFrame, S3bucket: String) = {
  dataFrame.map(row =>
      "text|label"
  ).coalesce(1).write.mode(SaveMode.Overwrite).text(S3bucket)
}

我们已经将其缩小到map函数，因为它在作为spark作业提交时工作：

def process(dataFrame: DataFrame, S3bucket: String) = {
  dataFrame.coalesce(1).write.mode(SaveMode.Overwrite).text(S3bucket)
}

有人知道是什么导致了这个问题吗？还有，我们怎么解决呢？我们被难倒了。

Answer 1

我认为，当worker试图访问一个SparkContext对象时，它会抛出一个NullPointerException，该对象只存在于驱动程序中，而不存在于worker中。

coalesce（）重新分区数据。当您只请求一个分区时，它将尝试在一个分区中压缩所有数据^*。这可能会给应用程序的内存占用带来很大压力。

一般情况下，最好不要只在1中收缩分区。

null