java - 如何知道 Apache Spark 中当前正在运行作业的哪个阶段？

Question

假设我在 Spark 中有一份工作；

CSV 文件 ==> 按列过滤 ==> 采样 ==> 另存为 JSON

现在我的要求是如何知道作业的哪个步骤(获取文件或过滤或采样)当前正在以编程方式执行(最好使用 Java API)？有什么办法吗？

我可以使用 SparkListener 类跟踪 Job、Stage 和 Task。它可以像跟踪阶段 ID 一样完成。但是如何知道作业链中的哪个阶段是哪个阶段的 Id。

当考虑按列过滤完成时，我想向用户发送通知。为此，我创建了一个扩展 SparkListener 类的类。但是我无法找到从哪里可以获得当前正在执行的转换名称的名称。是否有可能进行跟踪？

public class ProgressListener extends SparkListener{

  @Override
  public void onJobStart(SparkListenerJobStart jobStart)
  {

  }

  @Override
  public void onStageSubmitted(SparkListenerStageSubmitted stageSubmitted)
  {
      //System.out.println("Stage Name : "+stageSubmitted.stageInfo().getStatusString()); giving action name only
  }

  @Override
  public void onTaskStart(SparkListenerTaskStart taskStart)
  {
      //no such method like taskStart.name()
  }
}

Answer 1

您无法确切知道过滤操作何时开始或结束。

那是因为你有转换(filter,map,...)和 Action (count,foreach,...)。 Spark 会将尽可能多的操作放在一个阶段中。然后该阶段在输入的不同分区上并行执行。问题来了。

假设你有几个 worker 和下面的程序

LOAD ==> MAP ==> FILTER ==> GROUP BY + Aggregation

该程序可能有两个阶段:第一阶段将加载文件并应用map 和filter。然后输出将被打乱以创建组。在第二阶段将执行聚合。

现在的问题是，您有多个工作人员，每个工作人员将并行处理一部分输入数据。也就是说，集群中的每个执行程序都会收到一份程序(当前阶段)的副本，并在分配的分区上执行它。

您会看到，您将拥有多个并行执行的 map 和 filter 运算符实例，但不一定同时执行。在极端情况下，worker 1 将在 worker 20 开始之前完成阶段 1(因此在 worker 20 之前完成其 filter 操作)。

对于 RDD，Spark 使用 iterator model一个舞台里面。然而，对于最新 Spark 版本中的数据集，它们会在分区上创建一个循环并执行转换。这意味着在这种情况下，Spark 本身并不知道转换运算符何时完成单个任务!

长话短说:

1. 您无法知道阶段内的操作何时结束
2. 即使可以，也有多个实例将在不同时间完成。

所以，现在我已经遇到了同样的问题:

在我们的 Piglet project (请允许一些广告 ;-) )我们从 Pig Latin 脚本生成 Spark 代码并希望分析脚本。我最终在所有 用户操作符 之间插入了 mapPartition 操作符，它会将分区 ID 和当前时间发送到将评估消息的服务器。然而，这个解决方案也有其局限性……我还不完全满意。

但是，除非你能够修改程序，否则恐怕你无法实现你想要的。