apache-spark - Apache Spark 作业立即失败而无需重试，设置 maxFailures 不起作用

Question

我正在我的计算机上本地测试 Apache Spark 上的网络爬行/抓取程序。

该程序使用了一些 RDD 转换，这些转换采用了偶尔会失败的 volatile 函数。 (该函数的目的是将 URL 链接转换为网页，有时它调用的 headless 浏览器只是停电或过载 - 我无法避免)

我听说 Apache Spark 具有强大的故障转移和重试功能，任何不成功的转换或丢失的数据都可以从它可以找到的任何资源从头开始重新计算(听起来很神奇吧？)所以我没有在我的代码。

这是我的 Spark 配置:

val conf = new SparkConf().setAppName("MoreLinkedIn")
conf.setMaster("local[*]")
conf.setSparkHome(System.getenv("SPARK_HOME"))
conf.setJars(SparkContext.jarOfClass(this.getClass).toList)
conf.set("spark.task.maxFailures","40") //definitely enough

不幸的是，在大多数阶段和个别任务成功后，工作失败了。控制台的最新登录显示:

Exception in thread "main" org.apache.spark.SparkException: Job aborted due to stage failure: Task 1.0:7 failed 1 times, most recent failure: Exception failure in TID 23 on host localhost: org.openqa.selenium.TimeoutException: Timed out after 50 seconds waiting for...

看起来Spark在失败一次之后就懦弱地放弃了。我如何正确配置它以使其更加顽强？

(我的程序可以从 https://github.com/tribbloid/spookystuff 下载，抱歉代码/文档稀少和杂乱，我刚开始用几天)

ADD:如果你想自己尝试一下，下面的代码可以演示这个问题:

def main(args: Array[String]) {
val conf = new SparkConf().setAppName("Spark Pi")
conf.setMaster("local[*]")
conf.setSparkHome(System.getenv("SPARK_HOME"))
conf.setJars(SparkContext.jarOfClass(this.getClass).toList)
conf.set("spark.task.maxFailures","400000")
val sc = new SparkContext(conf)
val slices = if (args.length > 0) args(0).toInt else 8
val n = 100000 * slices
val count = sc.parallelize(1 to n, slices).map { i =>
  val x = java.lang.Math.random()
  if (x > 0.9) throw new IllegalStateException("the map has a chance of 10% to fail")
  x
}.reduce(_ + _)
sc.stop()
println("finished")
}

应该注意的是，在这篇文章中，同一个 IllegalStateException 被重试了 32 次:
Apache Spark Throws java.lang.IllegalStateException: unread block data

Answer 1

我知道这是一个非常古老的问题，但我遇到了完全相同的问题，并在寻找解决方案时遇到了这个问题。

有 3 种主 URL 格式可以在本地模式下提交 Spark 应用程序:

local - 一个线程(无并行性)，无重试

local[K] (或 local[*]) - 使用 K (或内核数)工作线程和集 task.maxFailures至 1 ( see here )

local[K, F] (或 local[*, F]) - 设置 task.maxFailures=F ，这就是我们所追求的。

咨询 Spark documentation详情。