apache-spark - Spark 流应用程序中的异常处理

Question

我开发了一个 Spark Streaming 应用程序，它检查文件流。我需要在任何驱动器异常时停止我的流媒体应用程序。我的代码如下:

val fileStream=..
  fileStream.checkpoint(Duration(batchIntervalSeconds * 1000 * 5))

//initiate the chekpointing
fileStream.foreachRDD(r=> {
  try {
    r.count()
  } catch {
    case ex: Exception => {
      ssc.stop(true, true)
    }

  }
}
)

但是，我从上面的代码中得到了异常

 yarn.ApplicationMaster: User class threw exception:
java.io.NotSerializableException: DStream checkpointing has been enabled but the DStreams with their functions are not serializable
org.apache.spark.streaming.StreamingContext
Serialization stack:
    - object not serializable (class: org.apache.spark.streaming.StreamingContext, value: org.apache.spark.streaming.StreamingContext@45ae9d8b)
    - field (class: UnionStream$$anonfun$creatingFunc$3, name: ssc$1, type: class org.apache.spark.streaming.StreamingContext)
    - object (class UnionStream$$anonfun$creatingFunc$3, <function1>)
    - field (class: org.apache.spark.streaming.dstream.DStream$$anonfun$foreachRDD$1$$anonfun$apply$mcV$sp$3, name: cleanedF$1, type: interface scala.Function1)
    - object (class org.apache.spark.streaming.dstream.DStream$$anonfun$foreachRDD$1$$anonfun$apply$mcV$sp$3, <function2>)
    - writeObject data (class: org.apache.spark.streaming.dstream.DStream)
    - object (class org.apache.spark.streaming.dstream.ForEachDStream, org.apache.spark.streaming.dstream.ForEachDStream@12481647)
    - writeObject data (class: org.apache.spark.streaming.dstream.DStreamCheckpointData)
    - object (class org.apache.spark.streaming.dstream.DStreamCheckpointData, 

我在 yarn 集群模式下运行我的代码..

Answer 1

要在 foreachRDD 调用内部发生异常时停止 spark 流应用程序，请不要尝试在 foreachRDD 内部捕获异常。而是将 ssc.awaitTermination 调用包装在 try/catch block 中并从那里调用 ssc.stop:

val ssc = createStreamingContext()        
ssc.start()
try {
  ssc.awaitTermination()
} catch {
  case e: Exception => 
    ssc.stop(stopSparkContext = true, stopGracefully = true)
    throw e // to exit with error condition
}