apache-spark - 在不知道作业 ID 的情况下如何获取 Apache Spark 作业进度？

Question

我有以下例子:

val sc: SparkContext // An existing SparkContext.
val sqlContext = new org.apache.spark.sql.SQLContext(sc)

val df = sqlContext.read.json("examples/src/main/resources/people.json")

df.count

我知道我可以使用 Spark 上下文使用 SparkListener 来监视作业；但是，这为我提供了所有作业的事件(我无法使用这些事件，因为我不知道作业 ID)。

如何才能仅获取“计数”操作的进度？

Answer 1

正如评论中已经建议的那样，可以使用 REST API of the Spark UI收集所需的数字。

主要问题是确定您感兴趣的阶段。从代码到阶段不存在 1:1 的映射。例如，单个计数将触发两个阶段(一个阶段用于计算数据帧每个分区中的元素，第二个阶段用于总结第一阶段的结果)。阶段通常获取触发其执行的操作的名称，尽管这可能是 changed within the code .

可以创建一个方法，查询 REST API 中具有特定名称的所有阶段，然后将这些阶段的所有任务数以及已完成的任务数相加。假设所有任务大约花费相似的执行时间(如果数据集有倾斜分区，则这种假设是错误的)，可以使用已完成任务的份额作为作业进度的衡量标准。

def countTasks(sparkUiUrl: String, stageName: String): (Int, Int) = {
  import scala.util.parsing.json._
  import scala.collection.mutable.ListBuffer
  def get(url: String) = scala.io.Source.fromURL(url).mkString

  //get the ids of all running applications and collect them in a ListBuffer
  val applications = JSON.parseFull(get(sparkUiUrl + "/api/v1/applications?staus=running"))
  val apps: ListBuffer[String] = new scala.collection.mutable.ListBuffer[String]
  applications match {
    case Some(l: List[Map[String, String]]) => l.foreach(apps += _ ("id"))
    case other => println("Unknown data structure while reading applications: " + other)
  }

  var countTasks: Int = 0;
  var countCompletedTasks: Int = 0;

  //get the stages for each application and sum up the number of tasks for each stage with the requested name
  apps.foreach(app => {
    val stages = JSON.parseFull(get(sparkUiUrl + "/api/v1/applications/" + app + "/stages"))
    stages match {
      case Some(l: List[Map[String, Any]]) => l.foreach(m => {
        if (m("name") == stageName) {
          countTasks += m("numTasks").asInstanceOf[Double].toInt
          countCompletedTasks += m("numCompleteTasks").asInstanceOf[Double].toInt
        }
      })
      case other => println("Unknown data structure while reading stages: " + other)
    }
  })

  //println(countCompletedTasks + " of " + countTasks + " tasks completed")
  (countTasks, countCompletedTasks)
}

针对给定的计数示例调用此函数

println(countTasks("http://localhost:4040", "show at CountExample.scala:16"))

将打印出两个数字:第一个是所有任务的数量，第二个是已完成任务的数量。

我已经使用 Spark 2.3.0 测试了此代码。在生产环境中使用它之前，肯定需要一些额外的修饰，尤其是一些更复杂的错误检查。不仅可以统计已完成的任务，还可以统计失败的任务，从而提高统计效果。