跨越两个作业的 Spark Sort 应用程序的 DAG答案

【问题标题】：DAG of Spark Sort application spanning two jobs跨越两个作业的 Spark Sort 应用程序的 DAG
【发布时间】：2022-04-23 22:25:18
【问题描述】：

我用 Spark 编写了一个非常简单的 Sort scala 程序。

object Sort {

    def main(args: Array[String]): Unit = {
        if (args.length < 2) {
            System.err.println("Usage: Sort <data_file> <save_file>" +
            " [<slices>]")
            System.exit(1)
        }


        val conf = new SparkConf().setAppName("BigDataBench Sort")
        val spark = new SparkContext(conf)
        val logger = new JobPropertiesLogger(spark,"/home/abrandon/log.csv")
        val filename = args(0)
        val save_file = args(1)
        var splits = spark.defaultMinPartitions
        if (args.length > 2){
            splits = args(2).toInt
        }
        val lines = spark.textFile(filename, splits)
        logger.start_timer()
        val data_map = lines.map(line => {
            (line, 1)
        })

        val result = data_map.sortByKey().map { line => line._1}
        logger.stop_timer()
        logger.write_log("Sort By Key: Sort App")
        result.saveAsTextFile(save_file)

        println("Result has been saved to: " + save_file)
    }



}

现在，我在想，由于只有一个广泛的转换（“sortByKey”），因此将跨越两个阶段。但是，我在 Job 0 中看到了两个工作，在 Job 0 中有一个阶段，在 Job 1 中有两个阶段。我错过了什么吗？我没有得到的是第二份工作的第一阶段。它似乎和 Job 0 的阶段做同样的工作。

【问题讨论】：

标签： scala apache-spark hadoop-yarn

【解决方案1】：

sortByKey 仅出现在第一个作业中。以下是第一个 Job 中的子任务：

spark.textFile（文件名，拆分）
data_map.sortByKey()

第二份工作有两个任务：

地图：.map { line => line._1}
saveAsTextFile

请注意，第二个作业中的第一个任务在第一个作业 (sortByKey) 离开的地方接手。您也可以通过查看执行时间来验证这一点：该阶段只需 4.4 分钟。

现在为什么第二份工作的可视化也显示了第一份工作中完成的所有工作？那个我不知道但是sortByKey 的工作似乎没有被执行两次。

【讨论】：

【解决方案2】：

当 RDD 处理排序转换时，它使用 RangePartitioner。要决定哪个键进入哪个分区，一个好主意是首先对源数据进行采样。因此，在第一项工作中，必须构建分区器。我猜是这样，但没有在源代码中严格确认。实际上，在早期的 MapReduce 框架中，通过采样构建范围分区器的想法非常普遍，并且由于其透明的编程模型而一目了然。

【讨论】：