停止 Spark Streaming：清洁线程中的异常，但它将继续运行

Question

我正在开发一个 Spark-Streaming 应用程序，我只是想获得一个 Kafka Direct Stream 工作的简单示例：

package com.username

import _root_.kafka.serializer.StringDecoder
import org.apache.spark.sql.SparkSession
import org.apache.spark.streaming.kafka._
import org.apache.spark.streaming.{Seconds, StreamingContext}

object MyApp extends App {
  val topic = args(0) // 1 topic
  val brokers = args(1) //localhost:9092
  val spark = SparkSession.builder().master("local[2]").getOrCreate()
  val sc = spark.sparkContext
  val ssc = new StreamingContext(sc, Seconds(1))
  val topicSet = topic.split(",").toSet
  val kafkaParams = Map[String, String]("metadata.broker.list" -> brokers)
  val directKafkaStream = KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder](ssc, kafkaParams, topicSet)

  // Just print out the data within the topic
  val parsers = directKafkaStream.map(v => v)
  parsers.print()

  ssc.start()
  val endTime = System.currentTimeMillis() + (5 * 1000) // 5 second loop

  while(System.currentTimeMillis() < endTime){
    //write something to the topic
    Thread.sleep(1000) // 1 second pause between iterations
  }

  ssc.stop()
}

这大部分都有效，无论我在 kafka 主题中写什么，它都会被包含在流批处理中并被打印出来。我唯一担心的是ssc.stop() 会发生什么：

dd/mm/yy hh:mm:ss WARN FileSystem: exception in the cleaner thread but it will continue to run
java.lang.InterruptException
    at java.lang.Object.wait(Native Method)
    at java.lang.ReferenceQueue.remove(ReferenceQueue.java:143)
    at java.lang.ReferenceQueue.remove(ReferenceQueue.java:164)
    at org.apache.hadoop.fs.FileSystem$Statistics$StatisticsDataReferenceCleaner.run(FileSystem.java:2989)
    at java.lang.Thread.run(Thread.java:748)

此异常不会导致我的应用程序失败或退出。我知道我可以将 ssc.stop() 包装到 try/catch 块中以抑制它，但是查看 API 文档让我相信这不是它的预期行为。我一直在网上寻找解决方案，但没有涉及 Spark 的任何内容提及此异常，我是否可以正确解决此问题？

Answer 1

我在直接使用 sbt run 启动进程时遇到了同样的问题。但是如果我将项目打包并以YOUR_SPARK_PATH/bin/spark-submit --class [classname] --master local[4] [package_path] 开头，它可以正常工作。希望这会有所帮助。