为什么sortBy转换会触发Spark作业？

Question

根据 Spark 文档，只有 RDD 操作可以触发 Spark 作业，并且在对其调用操作时会延迟评估转换。

我看到< code>sortBy转换函数被立即应用，并在SparkUI中显示为一个作业触发器。为什么？

Answer 1

根据Spark文档，Spark中只有动作触发作业，当对其调用动作时，转换被延迟评估。

总的来说，你是对的，但是正如你刚才所经历的，很少有例外，< code>sortBy就是其中之一(使用< code > ziptwithindex )。

事实上，它在Spark的JIRA中得到了报道，并以“不会修复”解决方案结束。请参阅SPARK-1021 sortByKey（）在不应该启动集群作业时启动集群作业。

您可以看到作业运行时启用了DAGScheduler日志记录（以及稍后在web UI中）：

scala> sc.parallelize(0 to 8).sortBy(identity)
INFO DAGScheduler: Got job 1 (sortBy at <console>:25) with 8 output partitions
INFO DAGScheduler: Final stage: ResultStage 1 (sortBy at <console>:25)
INFO DAGScheduler: Parents of final stage: List()
INFO DAGScheduler: Missing parents: List()
DEBUG DAGScheduler: submitStage(ResultStage 1)
DEBUG DAGScheduler: missing: List()
INFO DAGScheduler: Submitting ResultStage 1 (MapPartitionsRDD[4] at sortBy at <console>:25), which has no missing parents
DEBUG DAGScheduler: submitMissingTasks(ResultStage 1)
INFO DAGScheduler: Submitting 8 missing tasks from ResultStage 1 (MapPartitionsRDD[4] at sortBy at <console>:25)
DEBUG DAGScheduler: New pending partitions: Set(0, 1, 5, 2, 6, 3, 7, 4)
INFO DAGScheduler: ResultStage 1 (sortBy at <console>:25) finished in 0.013 s
DEBUG DAGScheduler: After removal of stage 1, remaining stages = 0
INFO DAGScheduler: Job 1 finished: sortBy at <console>:25, took 0.019755 s
res1: org.apache.spark.rdd.RDD[Int] = MapPartitionsRDD[6] at sortBy at <console>:25

Answer 2

< code>sortBy是使用< code>sortByKey实现的，它依赖于< code > RangePartitioner (JVM)或分区函数(Python)。当您调用< code > sort by /< code > sort by key 时，分区器(分区函数)会立即初始化，并对输入RDD进行采样以计算分区边界。你看到的工作对应于这个过程。

仅当您对新创建的RDD或其后代执行操作时，才会执行实际排序。