在哪些情况下会跳过 DAG 的阶段？答案

【问题标题】：In which situations are the stages of DAG skipped?在哪些情况下会跳过 DAG 的阶段？
【发布时间】：2019-10-14 14:19:05
【问题描述】：

我试图找出在我使用 RDD 的情况下 Spark 会跳过阶段的情况。我知道如果发生洗牌操作，它将跳过阶段。于是，我写了如下代码，看看是不是真的：

def main(args: Array[String]): Unit =
{
  val conf = new SparkConf().setMaster("local").setAppName("demo")
  val sc   = new SparkContext(conf)

  val d = sc.parallelize(0 until 1000000).map(i => (i%100000, i))

  val c=d.rightOuterJoin(d.reduceByKey(_+_)).collect
  val f=d.leftOuterJoin(d.reduceByKey(_+_)).collect
  val g=d.join(d.reduceByKey(_ + _)).collect
}

在检查 Spark UI 时，我得到了以下阶段的工作：

我希望跳过第 3 阶段和第 6 阶段，因为它们使用相同的 RDD 来计算所需的连接（考虑到在 shuffle 的情况下，spark 会自动缓存数据）。谁能解释一下为什么我在这里看不到任何跳过的阶段？以及如何修改代码以查看跳过的阶段？是否还有其他情况（除了洗牌）预计 Spark 会跳过阶段？

【问题讨论】：

stackoverflow.com/questions/34580662/…
@thebluephantom 该问题询问“跳过”阶段的含义。我的问题不同。正如我在问题中已经提到的以及您共享的链接所指出的那样，如果有洗牌，Spark 会缓存数据。但是在编写一个发生改组（加入）的简单代码时，我看不到任何跳过的阶段。
@thebluephantom 谢谢。如果您找到答案，请告诉我:)
@thebluephantom 谢谢！因此，我尝试了您的代码并将其与原始代码进行了比较。似乎只有在同一个 RDD 上应用超过 1 个操作（具有随机播放）时，Spark 才会自动跳过该阶段。
这就是我的观点，我会在午饭后发表。洗牌与缓存略有不同。这证明了这一点： val d = sc.parallelize(0 until 100000).map(i => (i%10000, i)).cache // 或不缓存，没关系 val c=d.rightOuterJoin(d. reduceByKey(+)) val f=d.leftOuterJoin(d.reduceByKey(+)) c.count c.collect // 跳过，打乱 f.count f.collect // 跳过，随机播放

标签： apache-spark rdd

【解决方案1】：

其实很简单。

在您的情况下，不能跳过任何内容，因为每个 Action 都有不同的 JOIN 类型。它需要扫描 d 和 d' 来计算结果。即使使用 .cache（您不使用它并且应该使用它来避免在每个操作上一直重新计算回源），这也没有什么区别。

看看这个简化版：

val d = sc.parallelize(0 until 100000).map(i => (i%10000, i)).cache // or not cached, does not matter

val c=d.rightOuterJoin(d.reduceByKey(_+_))
val f=d.leftOuterJoin(d.reduceByKey(_+_))

c.count
c.collect // skipped, shuffled 
f.count
f.collect // skipped, shuffled

显示此应用的以下工作：

(4) Spark Jobs
Job 116 View(Stages: 3/3)
Job 117 View(Stages: 1/1, 2 skipped)
Job 118 View(Stages: 3/3)
Job 119 View(Stages: 1/1, 2 skipped)

您可以看到，基于相同洗牌结果的连续动作会导致 val c 或 val f 的第二个动作/作业跳过一个或多个阶段。也就是说，c 和 f 的 join 类型是已知的，相同 join 类型的 2 个 Action 依次运行，从先前的工作中获益，即第二个 Action 可以依赖第一个 Action 的 shuffle，直接适用于 2nd行动。就这么简单。

【讨论】：