当我使用 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.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 WARN scheduler.TaskSetManager: Lost task 0.0 in 阶段 0.0（TID 0，cdh102，执行程序 2）：java.lang.NullPointerException 在 Recommend.DealLog$$anonfun$2$$anonfun$apply$1.apply(DealLog.scala:42) 在 Recommend.DealLog$$anonfun$2$$anonfun$apply$1.apply(DealLog.scala:41) 在 org.apache.spark.rdd.RDD$$anonfun$foreachPartition$1$$anonfun$apply$29.apply(RDD.scala:926) 在 org.apache.spark.rdd.RDD$$anonfun$foreachPartition$1$$anonfun$apply$29.apply(RDD.scala:926) 在 org.apache.spark.SparkContext$$anonfun$runJob$5.apply(SparkContext.scala:2071) 在 org.apache.spark.SparkContext$$anonfun$runJob$5.apply(SparkContext.scala:2071) 在 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$Worker.run(ThreadPoolExecutor.java:617) 在 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 INFO scheduler.TaskSetManager：在 cdh102 的阶段 0.0（TID 1）中丢失任务 0.1，执行程序 2： java.lang.NullPointerException (null) [重复 1] 19/07/26 18:21:56 INFO scheduler.TaskSetManager: 在阶段 0.0 启动任务 0.2 (TID 2, cdh102, executor 2, partition 0, PROCESS_LOCAL, 4706 字节） 19/07/26 18:21:56 INFO scheduler.TaskSetManager：在 cdh102 上的阶段 0.0（TID 2）丢失任务 0.2，执行程序 2： java.lang.NullPointerException (null) [重复 2] 19/07/26 18:21:56 INFO scheduler.TaskSetManager: 在阶段 0.0 启动任务 0.3 (TID 3, cdh102, executor 2, partition 0, PROCESS_LOCAL, 4706 字节） 19/07/26 18:21:56 INFO scheduler.TaskSetManager：在 cdh102 上的阶段 0.0（TID 3）丢失任务 0.3，执行程序 2： java.lang.NullPointerException (null) [重复 3] 19/07/26 18:21:56 ERROR scheduler.TaskSetManager: 阶段 0.0 中的任务 0 失败 4 次；中止工作 19/07/26 18:21:56 INFO cluster.YarnClusterScheduler：从池中删除了任务已全部完成的 TaskSet 0.0 19/07/26 18:21:56 INFO cluster.YarnClusterScheduler: 取消阶段 0 19/07/26 18:21:56 INFO scheduler.DAGScheduler: ResultStage 0 (foreachPartition at DealLog.scala:41) 由于 Job 在 1.092 秒内失败 由于阶段失败而中止：阶段 0.0 中的任务 0 失败了 4 次，大多数 最近失败：在 0.0 阶段丢失任务 0.3（TID 3，cdh102，执行程序 2): java.lang.NullPointerException 在 Recommend.DealLog$$anonfun$2$$anonfun$apply$1.apply(DealLog.scala:42) 在 Recommend.DealLog$$anonfun$2$$anonfun$apply$1.apply(DealLog.scala:41) 在 org.apache.spark.rdd.RDD$$anonfun$foreachPartition$1$$anonfun$apply$29.apply(RDD.scala:926) 在 org.apache.spark.rdd.RDD$$anonfun$foreachPartition$1$$anonfun$apply$29.apply(RDD.scala:926) 在 org.apache.spark.SparkContext$$anonfun$runJob$5.apply(SparkContext.scala:2071) 在 org.apache.spark.SparkContext$$anonfun$runJob$5.apply(SparkContext.scala:2071) 在 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$Worker.run(ThreadPoolExecutor.java:617) 在 java.lang.Thread.run(Thread.java:745)

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()
}