Scala-在提交jar时（而不是在shell中）触发NullPointerError

Question

我的spark作业引发了一个无法跟踪的空指针异常。当我打印潜在的null变量时，它们都填充在每个Worker上。我的数据不包含空值，因为同一作业在spark Shell中工作。下面是作业的execute函数，后面是错误消息。

没有在函数中定义的所有helper方法都是在spark作业对象的主体中定义的，所以我相信闭包不是问题所在。

override def execute(sc:SparkContext) = {
  def construct_query(targetTypes:List[String]) = Map("query" ->
    Map("nested" ->
      Map("path"->"annotations.entities.items",
        "query"-> Map("terms"->
          Map("annotations.entities.items.type"-> targetTypes)))))

  val sourceConfig = HashMap(
    "es.nodes" -> params.targetClientHost
  )

  // Base elastic search RDD returning articles which match the above query on entity types
  val rdd = EsSpark.esJsonRDD(sc,
    params.targetIndex,
    toJson(construct_query(params.entityTypes)),
    sourceConfig
  ).sample(false,params.sampleRate)

  // Mapping ES json into news article html" target="_blank">object, then extracting the entities list of
  // well defined annotations
  val objectsRDD = rdd.map(tuple => {
    val maybeArticle =
      try {
        Some(JavaJsonUtils.fromJson(tuple._2, classOf[SearchableNewsArticle]))
      }catch {
        case e: Exception => None
      }
    (tuple._1,maybeArticle)
  }
  ).filter(tuple => {tuple._2.isDefined && tuple._2.get.annotations.isDefined &&
    tuple._2.get.annotations.get.entities.isDefined}).map(tuple => (tuple._1, tuple._2.get.annotations.get.entities.get))

  // flat map the RDD of entities lists into a list of (entity text, (entity type, 1)) tuples
  (line 79) val entityDataMap: RDD[(String, (String, Int))] = objectsRDD.flatMap(tuple => tuple._2.items.collect({
    case item if (item.`type`.isDefined) && (item.text.isDefined) &&
   (line 81)(params.entityTypes.contains(item.`type`.get))  => (cleanUpText(item.text.get), (item.`type`.get, 1))
  }))

  // bucketize the tuples RDD into entity text, List(entity_type, entity_count) to make count aggregation and file writeouts
 // easier to follow
 val finalResults: Array[(String, (String, Int))] = entityDataMap.reduceByKey((x, y) => (x._1, x._2+y._2)).collect()

  val entityTypeMapping = Map(
    "HealthCondition" -> "HEALTH_CONDITION",
    "Drug" -> "DRUG",
    "FieldTerminology" -> "FIELD_TERMINOLOGY"
  )

  for (finalTuple <- finalResults) {
    val entityText = finalTuple._1
    val entityType = finalTuple._2._1
    if(entityTypeMapping.contains(entityType))
    {
                if(!Files.exists(Paths.get(entityTypeMapping.get(entityType).get+".txt"))){
        val myFile = new java.io.FileOutputStream(new   File(entityTypeMapping.get(entityType).get+".txt"),false)
        printToFile(myFile) {p => p.println(entityTypeMapping.get(entityType))}
      }
    }
    val myFile = new java.io.FileOutputStream(new   File(entityTypeMapping.get(entityType).get+".txt"),true)
    printToFile(myFile) {p => p.println(entityText)}
  }

}

以及下面的错误消息：

java.lang.NullPointerException在com.quid.gazetteers.gazetteergenerator$$anonfun$4$$anonfun$apply$1.isdefinedat（gazetteergenerator.scala:81)在com.quid.gazetteers.gazetteergenerator$$anonfun$4$$anonfun$apply$1.isdefinedat（gazetteergenerator.scala:79)在scala.collection.traversablelike$$anonfun$collection$1.apply(traversablelike.scala.immutable.list.foreach（list.scala:318)在scala.collection.traversablelike$class.collection VersableLike.Scala:278)在Scala.Collection.AbstractTraversable.Collect（Traversable.Scala:105)在com.quid.gazetteers.gazetteergenerator.Scala:79)在com.quid.gazetteers.gazetteergenerator$$anonfun$4。Apply（Gazetteergenerator.Scala:79)在scala.Collection.Iterator$$anon$13。HasNext(iterator.Scala:371)在org.apache.spark.util.Collection.ExternalSorter.InserTall（ExternalSorter.Scala:189)在org.apache.spark.shu。在org.apache.spark.scheduler.shufflem上写（sortshufflewriter.scala:64)aptask.runtask(shufflemaptask.scala:73)在org.apache.spark.scheduler.shufflemaptask.runtask(shufflemaptask.scala:41)在org.apache.spark.scheduler.task.run(task.scala:89)在org.apache.spark.executor.executor$taskrunner.run(executor.scala:214)在java.util.concurrent.threadpoolexecutor.runworker(threadpoolexecutor.1142)在java.util.concurrent.threadpoolexecutor$worker.run(threadpoolexecutor.617)在java.lang.thread.run（thread.java：745）

Answer 1

这个问题已经解决了。params属性未序列化，且对spark Workers可用。解决方案是在需要params属性的区域范围内形成一个spark广播变量。