我有一个 HadoopRDD，我从中创建第一个带有简单 Map 函数的 RDD，然后从第一个 RDD 中创建第二个 RDD，并带有另一个简单的 Map 函数。类似的东西：
HadoopRDD -> RDD1 -> RDD2。
我的问题是 Spak 是否会逐个记录迭代 HadoopRDD 记录以生成 RDD1，然后它会逐个记录迭代 RDD1 记录以生成 RDD2，还是它会遍历 HadoopRDD，然后一次生成 RDD1，然后生成 RDD2。
【问题讨论】：
标签： apache-spark bigdata rdd
简答：rdd.map(f).map(g) 将一次性执行。
Spark 将作业拆分为多个阶段。应用于数据分区的阶段是一项任务。
在一个阶段中，Spark 会尝试将尽可能多的操作流水线化。 “可能”取决于重新排列数据的需要：需要 shuffle 的操作通常会破坏管道并创建新阶段。
实际上：
Given `rdd.map(...).map(..).filter(...).sort(...).map(...)`
将分为两个阶段：
.map(...).map(..).filter(...)
.sort(...).map(...)
这可以使用 rdd.toDebugString 从 rdd 中检索到
上面的相同作业示例将产生以下输出：
val mapped = rdd.map(identity).map(identity).filter(_>0).sortBy(x=>x).map(identity)
scala> mapped.toDebugString
res0: String =
(6) MappedRDD[9] at map at <console>:14 []
| MappedRDD[8] at sortBy at <console>:14 []
| ShuffledRDD[7] at sortBy at <console>:14 []
+-(8) MappedRDD[4] at sortBy at <console>:14 []
| FilteredRDD[3] at filter at <console>:14 []
| MappedRDD[2] at map at <console>:14 []
| MappedRDD[1] at map at <console>:14 []
| ParallelCollectionRDD[0] at parallelize at <console>:12 []
现在，进入您问题的关键点：流水线非常有效。完整的管道将应用于每个分区的每个元素一次。这意味着rdd.map(f).map(g) 的执行速度将与rdd.map(f andThen g) 一样快（有一些可忽略的开销）
【讨论】：
Apache Spark 将不按特定顺序逐个记录迭代 HadoopRDD 记录（数据将被拆分并发送给工作人员），并“应用”第一个转换来计算 RDD1。之后，对 RDD1 的每个元素应用第二次变换得到 RDD2，同样没有特定的顺序，以此类推进行连续变换。您可以从map 方法签名中注意到它：
// Return a new RDD by applying a function to all elements of this RDD.
def map[U](f: (T) ⇒ U)(implicit arg0: ClassTag[U]): RDD[U]
Apache Spark 遵循 DAG（有向无环图）执行引擎。在需要值之前它实际上不会触发任何计算，因此您必须区分transformations 和actions。
编辑：
在性能方面，我并不完全了解 Spark 的底层实现，但我知道除了在相关阶段添加额外（不必要的）任务之外，不应该有明显的性能损失。根据我的经验，您通常不会连续使用相同“性质”的转换（在本例中是两个连续的map）。当进行洗牌操作时，您应该更加关注性能，因为您正在移动数据，这对您的工作绩效有明显的影响。 Here 你可以找到一个常见的问题。
【讨论】：