为什么 Spark 不并行化 randomSplit答案

【问题标题】：Why doesn't Spark parallelize randomSplit为什么 Spark 不并行化 randomSplit
【发布时间】：2017-02-05 19:14:30
【问题描述】：

我有一些创建一系列 RDD 的 Spark 代码。最后，我调用randomSplit 将其分成3 组，然后将每组写入磁盘。所以第一阶段是：

获取一些数据
做一些转换
缓存结果
通过randomSplit拆分
将所有拆分写入磁盘

因为步骤 (4) 将事物分成 3 组，所以这里有 3 个不同的 Spark 阶段。在第一阶段结束时，我们开始用完第一阶段的任务，但有可用的执行器：

此时已经计算了几个分区的数据集。据我了解randomSplit 在逐个分区的基础上运行；换句话说，它不需要洗牌或收集 - 它只是在每个分区的基础上随机选择行。如果这是正确的，那么第 2 阶段的某些任务就没有理由无法在可用的执行程序上运行——它们的 RDD 的分区已被计算和缓存。为什么 Spark 不启动某些第 2 阶段任务以利用可用资源。

注意：显然，“他们可以但他们没有”的答案是完全有效的。我想我真正要问的是，是否有一些我没有想到的技术原因导致这不可能（或非常困难），或者这只是实施过程中的疏忽？

这是代码的简化版本（在 Kotlin 中）：

fun run(sc: JavaSparkContext, opts: Options) {
    val allData = fetchABunchOfData()

    val allDataRdd = sc.parallelize(allData)

    val taggedAndTokenized = allDataRdd.mapPartitions { addTagsAndTokens(it) }

    // Convert each ResponseData to a JSON String
    val jsonStrings = taggedAndTokenized.map {
        val mapper = AnxJsonUtils.getMapper()
        mapper.writeValueAsString(it)
    }

    // the randomSplit below creates 3 distinct RDD lineags so if we don't cache the parsing stuff we'll parse the
    // entire document set twice.
    jsonStrings.cache()
    val trainValidTest =
            jsonStrings.randomSplit(doubleArrayOf(opts.trainFrac, opts.validFrac, opts.testFrac), splitSeed)
    trainValidTest[0].saveAsTextFile(opts.outPath + "/" + TRAIN_NAME)
    trainValidTest[1].saveAsTextFile(opts.outPath + "/" + VALID_NAME)
    trainValidTest[2].saveAsTextFile(opts.outPath + "/" + TEST_NAME)
}

【问题讨论】：

澄清一下，您是在问为什么第二个saveAsTextFile 在第一个完成之前没有开始？

标签： apache-spark

【解决方案1】：

出于多种原因，saveAsTextFile 是一个阻塞调用。这意味着 Spark master 在第一个保存指令完成之前不会收到第二个保存指令。

也就是说，如果您希望利用这些可用资源，您可以做的是在三个单独的线程中调用 saveAsTextFile 并等待它们的未来。一旦工作人员完成了第一个任务的分区，它就可以开始第二个任务。

【讨论】：