Spark Streaming异常：java.util.NoSuchElementException：None.get

Question

我正在将数据转换为数据帧，将其写入HDFS：

密码

object KafkaSparkHdfs {

  val sparkConf = new SparkConf().setMaster("local[*]").setAppName("SparkKafka")
  sparkConf.set("spark.driver.allowMultipleContexts", "true");
  val sc = new SparkContext(sparkConf)

  def main(args: Array[String]): Unit = {
    val sqlContext = new org.apache.spark.sql.SQLContext(sc)
    import sqlContext.implicits._

    val ssc = new StreamingContext(sparkConf, Seconds(20))

    val kafkaParams = Map[String, Object](
      "bootstrap.servers" -> "localhost:9092",
      "key.deserializer" -> classOf[StringDeserializer],
      "value.deserializer" -> classOf[StringDeserializer],
      "group.id" -> "stream3",
      "auto.offset.reset" -> "latest",
      "enable.auto.commit" -> (false: java.lang.Boolean)
    )

    val topics = Array("fridaydata")
    val stream = KafkaUtils.createDirectStream[String, String](
      ssc, PreferConsistent, Subscribe[String, String](topics, kafkaParams)
    )

    val lines = stream.map(consumerRecord => consumerRecord.value)
    val words = lines.flatMap(_.split(" "))
    val wordMap = words.map(word => (word, 1))
    val wordCount = wordMap.reduceByKey(_ + _)

    wordCount.foreachRDD(rdd => {
      val dataframe = rdd.toDF(); 
      dataframe.write
        .mode(SaveMode.Append)
        .save("hdfs://localhost:9000/newfile24")     
    })

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

已创建文件夹，但未写入文件。

程序因以下错误而终止：

    18/06/22 16:14:41 ERROR Executor: Exception in task 0.0 in stage 0.0 (TID 0)
    java.util.NoSuchElementException: None.get
    at scala.None$.get(Option.scala:347)
    at scala.None$.get(Option.scala:345)
    at org.apache.spark.storage.BlockInfoManager.releaseAllLocksForTask(BlockInfoManager.scala:343)
    at org.apache.spark.storage.BlockManager.releaseAllLocksForTask(BlockManager.scala:670)
    at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:289)
    at java.lang.Thread.run(Thread.java:748)
    18/06/22 16:14:41 WARN TaskSetManager: Lost task 0.0 in stage 0.0 (TID 0, localhost, executor driver): java.util.NoSuchElementException: None.get
    at scala.None$.get(Option.scala:347)
    at scala.None$.get(Option.scala:345)
    at org.apache.spark.storage.BlockInfoManager.releaseAllLocksForTask(BlockInfoManager.scala:343)
    at org.apache.spark.storage.BlockManager.releaseAllLocksForTask(BlockManager.scala:670)
    at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:289)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)

在我的pom中，我使用了各自的依赖项：

• 火花-core_2.11
• 火花-sql_2.11
• 火花-streaming_2.11
• 火花流-kafka-0-10_2.11

Answer 1

匿名用户

这里有一个明显的问题-合并（1）。

dataframe.coalesce(1)

虽然在许多情况下减少文件数量可能很诱人，但当且仅当数据量足够低以供节点处理时才应该这样做（显然它不在这里）。

此外，让我引述留档：

但是，如果要进行剧烈合并，例如，将numPartitions=1，这可能会导致计算在比您喜欢的节点更少的节点上进行（例如，在numPartitions=1的情况下为一个节点）。为了避免这种情况，可以调用重新分区。这将添加一个洗牌步骤，但意味着当前上游分区将并行执行（无论当前分区是什么）。

结论是，您应该根据预期的数据量和所需的并行性相应地调整参数<这样的代码>合并（1）在实践中很少有用，尤其是在流式处理这样的上下文中，数据属性可能随时间而不同。

Answer 2

错误是由于试图同时运行多个spark上下文。将allowMultipleContexts设置为true主要用于测试目的，不鼓励使用。因此，问题的解决方案是确保在任何地方都使用相同的SparkContext。从代码中我们可以看到，SparkContext（sc）用于创建一个SQLContext，这很好。但是，在创建StreamingContext时，不使用它，而是使用SparkConf。

通过查看留档，我们可以看到：

通过提供新SparkContext所需的配置来创建StreamingContext

换句话说，通过使用SparkConf作为参数，将创建一个新的SparkContext。现在有两个单独的上下文。

这里最简单的解决方案是继续使用与之前相同的上下文。将创建StreamingContext的行更改为：

val ssc = new StreamingContext(sc, Seconds(20))

注意：在较新版本的Spark（2.0）中，请改用SparkSession。然后可以使用StreamingContext（spark.sparkContext，…）创建新的流上下文 。它可以如下所示：

val spark = SparkSession().builder
  .setMaster("local[*]")
  .setAppName("SparkKafka")
  .getOrCreate()

import sqlContext.implicits._
val ssc = new StreamingContext(spark.sparkContext, Seconds(20))