当我使用SparkStreaming处理Kafka的消息时，我得到了NullPointerException

Question

我正在为Kafka和SparkStreaming编写一些代码，当我将它们放在Yarn-Cluster上时，它报告了NullPointerException。

但它在我的电脑上运行良好（独立模式）

那它有什么问题呢？

//这是代码

import java.util.Properties

import org.apache.kafka.common.serialization.StringDeserializer
import org.apache.log4j.Logger
import org.apache.spark.html" target="_blank">sql.SparkSession
import org.apache.spark.streaming.kafka010.ConsumerStrategies.Subscribe
import org.apache.spark.streaming.kafka010.KafkaUtils
import org.apache.spark.streaming.kafka010.LocationStrategies.PreferConsistent
import org.apache.spark.streaming.{Seconds, StreamingContext}

object DealLog extends App {

  val spark=SparkSession.builder().appName(" DealLog").getOrCreate()
  val sc = spark.sparkContext
  val ssc: StreamingContext= new StreamingContext(sc, Seconds(3))

  val log = Logger.getLogger(this.getClass)
  val pro = new Properties()
  val in = Thread.currentThread().getContextClassLoader.getResourceAsStream("config.properties")
  pro.load(in)
  //  ssc.checkpoint("hdfs://192.168.0.240:8022/bigdata/checkpoint2")
  val bootstrap=pro.getProperty("kafka.brokers")
  val kafkaParams = Map[String, Object]("bootstrap.servers" -> bootstrap,
    "key.deserializer" -> classOf[StringDeserializer],
    "value.deserializer" -> classOf[StringDeserializer],
    "group.id" -> "userlabel",
    "auto.offset.reset" -> "latest",
    "enable.auto.commit" -> (true: java.lang.Boolean)
  )
  val topicsSet = Array(pro.getProperty("kafkaconsume.topic"))
  val ds = KafkaUtils.createDirectStream[String,String](
    ssc,
    PreferConsistent,
    Subscribe[String,String](topicsSet,kafkaParams)
  ).map(s=>{(s.value())})

  ds.foreachRDD(p=>{
    log.info("ds.foreachRdd p=="+ p)
    p.foreachPartition(per=>{
      log.info("per-------"+per)
      per.foreach(rdd=> {
        log.info("rdd---------"+ rdd)
        if(rdd.isEmpty){
          log.info("null ")
        }
        else{
          log.info("not null..")
        }
        log.info("complete")

      })
    })
  })
  ssc.start()
  ssc.awaitTermination()
}

这里例外-----------------------------------

19/07/26 18:21:56警告Scheduler.TaskSetManager：在stage 0.0中丢失任务0.0（TID 0，cdh102，执行器2）：java.lang.NullPointerException at Recomment.Deallog$$AnonFun$2$$AnonFun$Apply$1.Apply(Deallog.scala:42)at Recomment.Deallog$$AnonFun$2$$AnonFun$Apply$1.Apply(Deallog.scala:41)at org.apache.spark.rdd.rdd$$AnonFun$ForeachPartition在java.lang.Thread.Run（thread.java:745)

19/07/26 18:21:56 INFO scheduler.TaskSetManager: Starting task 0.1 in stage 0.0 (TID 1, cdh102, executor 2, partition 0, PROCESS_LOCAL,

4706字节）19/07/26 18:21:56信息调度器.TaskSetManager：丢失了cdh102上阶段0.0（TID 1）中的任务0.1,executor 2:java.lang.NullPointerException（null）[duplicate 1]19/07/26 18:21:56 INFO Scheduler.TaskSetManager：在stage 0.0（TID 2,cdh102,executor 2,partition 0,PROCESS_LOCAL,4706字节）中启动任务0.2 19/07/26 18:21:56 INFO Scheduler.TaskSetManager：在stage 0.0（TID 2,cdh102,executor 2,partition 0,PROCESS_LOCAL,4706字节）中丢失任务0.2,cdh102,executor 2,executor 2,executor 2,executor正在中止作业19/07/26 18:21:56信息集群。YarnClusterScheduler：已从池19/07/26 18:21:56信息集群中删除任务已全部完成的TaskSet 0.0。YarnClusterScheduler：取消阶段0 19/07/26 18:21:56信息调度器。DagScheduler：由于阶段失败导致作业中止而导致ResultStage 0（foreachPartition在Deallog.Scala:41)在1.092s中失败：阶段0.0中的任务0失败了4次，最近的失败：丢失了阶段0.0中的任务0.3（TID 3,cdh102,executor 2）：java.lang.NullPointerException（建议）。71）在org.apache.spark.scheduler.resulttask.runtask（resulttask.scala：87）在org.apache.spark.scheduler.task.run（task.scala：109）在org.apache.spark.executor.executor$taskrunner.run（executor.scala：338）在java.util.concurrent.ThreadPoolExecutor.runworker（ThreadPoolExecutor.java:1142）在java.util.concurrent.ThreadPoolExecutor

Answer 1

我想你的问题可能来自这条线

if(rdd.isEmpty)

因为您编写代码的方式实际上并不是RDD。在调用foreachPartition之后，您将获得该分区的迭代器。然后，当您在该迭代器上调用foreach时，您将访问该分区迭代器上的实际记录。所以在这一行上处理的是来自dStream的记录。因此，您可能会对抛出该异常的空字符串/值调用.isempty。

可以将.isempty替换为

if(record == null)

但你不必那么做。您可以只检查RDD本身是否为空。你能试试下面的吗？

import org.apache.kafka.common.serialization.StringDeserializer
import org.apache.log4j.Logger
import org.apache.spark.sql.SparkSession
import org.apache.spark.streaming.kafka010.ConsumerStrategies.Subscribe
import org.apache.spark.streaming.kafka010.KafkaUtils
import org.apache.spark.streaming.kafka010.LocationStrategies.PreferConsistent
import org.apache.spark.streaming.{Seconds, StreamingContext}

object DealLog extends App {

  val spark = SparkSession.builder().appName(" DealLog").getOrCreate()
  val sc = spark.sparkContext
  val ssc: StreamingContext = new StreamingContext(sc, Seconds(3))

  val log = Logger.getLogger(this.getClass)
  val pro = new Properties()
  val in = Thread.currentThread().getContextClassLoader.getResourceAsStream("config.properties")
  pro.load(in)
  //  ssc.checkpoint("hdfs://192.168.0.240:8022/bigdata/checkpoint2")
  val bootstrap = pro.getProperty("kafka.brokers")
  val kafkaParams = Map[String, Object]("bootstrap.servers" -> bootstrap,
    "key.deserializer" -> classOf[StringDeserializer],
    "value.deserializer" -> classOf[StringDeserializer],
    "group.id" -> "userlabel",
    "auto.offset.reset" -> "latest",
    "enable.auto.commit" -> (true: java.lang.Boolean)
  )
  val topicsSet = Array(pro.getProperty("kafkaconsume.topic"))
  val ds = KafkaUtils.createDirectStream[String, String](
    ssc,
    PreferConsistent,
    Subscribe[String, String](topicsSet, kafkaParams)
  ).map(s => {
    (s.value())
  })

  ds.foreachRDD(rdd => {
    log.info("ds.foreachRdd p==" + rdd)
    if (!rdd.isEmpty) {
      rdd.foreachPartition(partition => {
        log.info("per-------" + partition)
        partition.foreach(record => {
          log.info("record---------" + record)
        })
      })
    } else log.info("rdd was empty")

    log.info("complete")
  })
  ssc.start()
  ssc.awaitTermination()
  ssc.stop()
}