scala - "Immortal"Spark 流作业？

Question

好的，所以我问了一个有点相似的 question与 Spark 如何在内部处理异常有关，但我当时的示例并不是很清楚或完整。那里的答案为我指明了方向，但我无法真正解释一些事情。

我已经设置了一个虚拟的 Spark 流应用程序，在转换阶段我有一个俄罗斯轮盘赌表达式，它可能会或不会抛出异常。如果抛出异常，我会停止 Spark 流上下文。就是这样，没有其他逻辑，没有RDD转型。

object ImmortalStreamingJob extends App {
  val conf = new SparkConf().setAppName("fun-spark").setMaster("local[*]")
  val ssc  = new StreamingContext(conf, Seconds(1))

  val elems = (1 to 1000).grouped(10)
    .map(seq => ssc.sparkContext.parallelize(seq))
    .toSeq
  val stream = ssc.queueStream(mutable.Queue[RDD[Int]](elems: _*))

  val transformed = stream.transform { rdd =>
    try {
      if (Random.nextInt(6) == 5) throw new RuntimeException("boom")
      else println("lucky bastard")
      rdd
    } catch {
      case e: Throwable =>
        println("stopping streaming context", e)
        ssc.stop(stopSparkContext = true, stopGracefully = false)
        throw e
    }
  }

  transformed.foreachRDD { rdd =>
    println(rdd.collect().mkString(","))
  }

  ssc.start()
  ssc.awaitTermination()
}

在 IntelliJ 中运行它会在某个时候抛出异常。有趣的部分:

如果在第一个转换中抛出异常(在处理第一个 RDD 时)，spark 上下文将停止并且应用程序终止，这就是我想要的

如果在至少一个 RDD 之后抛出异常已处理，打印错误信息后app挂起，永不停止，这不是我想要的

为什么应用程序在第二种情况下挂起而不是死掉？

我在 Scala 2.11.8 上运行 Spark 2.1.0。退出 try-catch 解决了问题(Spark 自行停止)。另外，移出内部的 try-catch foreachRDD解决了这个问题。

但是，我正在寻找可以帮助我了解此特定示例中发生的情况的答案。

Answer 1

你只会看到异常表现在 Action 中(如本例中的 foreachRDD)而不是转换(如本例中的 transform)，因为 Action 执行转换是惰性的。这意味着您的转换甚至在操作之前都不会发生。这是必要的原因需要改变您对分布式处理如何工作的心理模型。

考虑一个传统的单线程程序。代码逐行执行，如果抛出异常但未处理，则后续代码行不会执行。在分布式系统中，相同的 Spark 转换在多台机器上并行运行(以不同的速度)，当抛出异常时会发生什么？这不是那么简单，因为一台机器上的异常独立于其他机器上的代码处理，这就是你想要的。希望分散在整个集群中的所有独立任务只是在出现异常时关闭，这只是单机思维，不会转化为分布式范式。司机应该如何处理？

根据现在 Databricks 和 Spark 的创建者之一回到伯克利的 Matei Zaharia 所说，“异常应该发送回驱动程序并记录在那里(如果任务失败超过 4 次，则会抛出 SparkException)。” (顺便说一下，这个默认的重试次数可以用 spark.task.maxFailures 更改。)。所以如果在执行器上正确配置了 Log4J，异常就会被记录在那里；然后它将被序列化并发回给驱动程序，默认情况下，驱动程序将再试 3 次。

在您的特定情况下，我猜您有几件事情要做。首先，您在单台机器上运行，这会让人误解异常处理在分布式模型中的工作方式。其次，您过早地停止了上下文。停止上下文 is an extremely destructive operation ，包括停止所有听众和 DAGScheduler .坦率地说，我不知道当你基本上关掉灯时，你怎么能期望 Spark 把所有东西都包装得如此整洁。

最后，我要提到一个更优雅的异常处理模型可能会在 Try 中执行您的转换。 .您最终可能会得到更麻烦的代码，因为您的转换将返回 RDD[Try[T]]或 DStream[Try[T]] ，这意味着您必须处理 Success和 Failure每个元素的案例。但是您将能够利用 monad 提供的所有好处向下游传播成功和错误信息，包括映射 RDD[Try[A]] => RDD[Try[B]]甚至使用 for理解(凭借 flatMap )。