在Spark中阶段是如何拆分为任务的？

Question

下面我们假设每个时间点只有一个Spark作业在运行。

以下是我所理解的在Spark中发生的事情：

1. 创建sparkcontext时，每个工作节点都启动一个执行器。执行程序是单独的进程(JVM)，它连接回驱动程序。每个执行器都有驱动程序的jar。退出驱动程序，关闭执行程序。每个执行程序可以保存一些分区。
2. 执行作业时，将根据沿袭图创建执行计划。
3. 执行作业被拆分为多个阶段，其中阶段包含相同数量的相邻转换和操作（在沿袭图中），但不包含洗牌。这样，阶段就被洗牌分开了。

null

但是

我如何将阶段划分为那些任务？

具体地说：

null

那是正确的吗？但是即使那是正确的，我上面的问题也没有全部得到回答，因为它仍然是开放的，多个操作（例如多个映射）是在一个任务内，还是每个操作被分离成一个任务。

火花中的任务是什么？Spark worker如何执行jar文件？Apache Spark调度器如何将文件拆分为任务？但我觉得我的问题没有得到清楚的回答。

Answer 1

你的轮廓很不错。回答你的问题

• 需要为每个阶段的每个数据分区启动单独的任务。考虑每个分区可能驻留在不同的物理位置上-例如HDFS中的块或本地文件系统的目录/卷。

请注意，stages的提交是由DAG调度程序驱动的。这意味着不相互依赖的阶段可能会被提交到集群中并行执行：这使得集群上的并行化能力最大化。因此，如果我们的数据流中的操作可以同时发生，我们将期望看到多个阶段启动。

null

• 用于并行加载两个数据源的各1级=2级
• 表示连接的第三阶段，它依赖于其他两个阶段
• 注意：所有对连接的数据进行的后续操作都可能在同一阶段中执行，因为它们必须顺序发生。启动其他阶段没有任何好处，因为在完成前一个操作之前，这些阶段无法开始工作。

这是玩具程序

val sfi  = sc.textFile("/data/blah/input").map{ x => val xi = x.toInt; (xi,xi*xi) }
val sp = sc.parallelize{ (0 until 1000).map{ x => (x,x * x+1) }}
val spj = sfi.join(sp)
val sm = spj.mapPartitions{ iter => iter.map{ case (k,(v1,v2)) => (k, v1+v2) }}
val sf = sm.filter{ case (k,v) => v % 10 == 0 }
sf.saveAsTextFile("/data/blah/out")

这是结果的DAG