apache-spark - Spark - 如何通过 'SparkLauncher' 识别失败的作业

Question

我使用的是 Spark 2.0，有时我的工作会因为输入问题而失败。例如，我正在根据日期从 S3 文件夹中读取 CSV 文件，如果当前日期没有数据，则我的工作无需处理，因此会引发如下异常。这会打印在驱动程序的日志中。

Exception in thread "main" org.apache.spark.sql.AnalysisException: Path does not exist: s3n://data/2016-08-31/*.csv;
    at org.apache.spark.sql.catalyst.analysis.CheckAnalysis$class.failAnalysis(CheckAnalysis.scala:40)
    at org.apache.spark.sql.catalyst.analysis.Analyzer.failAnalysis(Analyzer.scala:58)
    at org.apache.spark.sql.catalyst.analysis.CheckAnalysis$$anonfun$checkAnalysis$1.apply(CheckAnalysis.scala:174)
    at org.apache.spark.sql.catalyst.analysis.CheckAnalysis$$anonfun$checkAnalysis$1.apply(CheckAnalysis.scala:67)
...
    at org.apache.spark.deploy.SparkSubmit$.org$apache$spark$deploy$SparkSubmit$$runMain(SparkSubmit.scala:729)
    at org.apache.spark.deploy.SparkSubmit$.doRunMain$1(SparkSubmit.scala:185)
    at org.apache.spark.deploy.SparkSubmit$.submit(SparkSubmit.scala:210)
    at org.apache.spark.deploy.SparkSubmit$.main(SparkSubmit.scala:124)
    at org.apache.spark.deploy.SparkSubmit.main(SparkSubmit.scala)
 16/09/03 10:51:54 INFO SparkContext: Invoking stop() from shutdown hook
 16/09/03 10:51:54 INFO SparkUI: Stopped Spark web UI at http://192.168.1.33:4040
 16/09/03 10:51:54 INFO StandaloneSchedulerBackend: Shutting down all executors
 16/09/03 10:51:54 INFO CoarseGrainedSchedulerBackend$DriverEndpoint: Asking each executor to shut down
Spark App app-20160903105040-0007 state changed to FINISHED

然而，尽管有这个未捕获的异常，我的 Spark 作业状态是“完成”。我希望它处于“失败”状态，因为有一个异常(exception)。为什么它被标记为FINISHED？我如何知道作业是否失败？

注意:我正在使用 SparkLauncher 生成 Spark 作业，并通过 AppHandle 监听状态更改。但是我收到的状态更改已完成，而我期待失败。

Answer 1

一个成品 你看到的是 Spark 应用程序，而不是一份工作。是成品 因为 Spark 上下文能够正确启动和停止。

您可以使用 查看任何工作信息。 JavaSparkStatusTracker .
对于事件作业，无需进行任何额外操作，因为它具有“.getActiveJobIds”方法。

为了完成/失败，您需要在调用 spark 执行的线程中设置作业组 ID:

JavaSparkContext sc;
... 
sc.setJobGroup(MY_JOB_ID, "Some description");

然后，无论何时需要，您都可以读取指定作业组中每个作业的状态:

JavaSparkStatusTracker statusTracker = sc.statusTracker();
for (int jobId : statusTracker.getJobIdsForGroup(JOB_GROUP_ALL)) {
    final SparkJobInfo jobInfo = statusTracker.getJobInfo(jobId);
    final JobExecutionStatus status = jobInfo.status();
}

JobExecutionStatus 可以是 之一正在运行 , 成功 , 失败 , 未知 ;最后一个用于提交作业但未实际启动的情况。

注:所有这些都可以从 Spark 驱动程序中获得，它是您使用 启动的 jar 包。 SparkLauncher .所以上面的代码应该放入jar中。

如果您想从 方面检查一般情况下是否有任何故障 Spark 发射器 ，您可以使用类型 退出由 Jar 启动的应用程序，退出代码不同于 0 System.exit(1) , 如果检测到作业失败。 进程返回者 SparkLauncher::启动包含 退出值 方法，因此您可以检测它是否失败。