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

Question

让我们假设以下每个时间点只运行一个Spark作业.

到目前为止我得到了什么

以下是我了解Spark中发生的事情:

1. 当SparkContext被创建,每个工作节点开始执行人.执行程序是单独的进程(JVM),它连接回驱动程序.每个执行程序都有驱动程序的jar.退出驱动程序,关闭执行程序.每个执行程序都可以保存一些分区.
2. 执行作业时,将根据沿袭图创建执行计划.
3. 执行作业分为几个阶段,其中阶段包含尽可能多的邻近(在谱系图中)转换和动作,但没有随机播放.因此阶段通过混洗分开.

我明白那个

• 任务是通过序列化Function对象从驱动程序发送到执行程序的命令.
• 执行程序反序列化(使用驱动程序jar)命令(任务)并在分区上执行它.

但

问题(S)

如何将舞台分成这些任务？

特别:

1. 是由变换和操作决定的任务还是可以在任务中进行多个转换/操作？
2. 任务是由分区确定的(例如,每个分区每个阶段一个任务).
3. 任务是由节点确定的(例如,每个节点每个阶段一个任务)？

我的想法(只有部分答案,即使是正确的)

在https://0x0fff.com/spark-architecture-shuffle中,随着图像解释了随机播放

我得到了规则的印象

每个阶段被分成#count-of-partitions任务,不考虑节点数量

对于我的第一张图片,我会说我有3个地图任务和3个减少任务.

对于来自0x0fff的图像,我会说有8个地图任务和3个减少任务(假设只有三个橙色和三个深绿色文件).

无论如何都要打开问题

那是对的吗？但即使这是正确的,我上面的问题也没有全部回答,因为它仍然是开放的,无论多个操作(例如多个地图)是在一个任务内还是每个操作分成一个任务.

别人怎么说

Spark的任务是什么？Spark工作人员如何执行jar文件？以及Apache Spark调度程序如何将文件拆分为任务？是相似的,但我觉得我的问题在那里得不到清楚.

Answer 1

你有一个非常好的大纲.回答你的问题

• 单独task 确实需要推出针对每个数据的每个分区stage.考虑每个分区可能位于不同的物理位置 - 例如HDFS中的块或本地文件系统的目录/卷.

请注意,Stages 的提交是由DAG Scheduler.这意味着可以将不相互依赖的阶段提交到集群以便并行执行:这最大化了集群上的并行化功能.因此,如果我们的数据流中的操作可以同时发生,我们将期望看到启动多个阶段.

我们可以看到以下玩具示例中的操作,其中我们执行以下类型的操作:

• 加载两个数据源
• 分别对两个数据源执行一些映射操作
• 加入他们
• 对结果执行一些映射和过滤操作
• 保存结果

那么我们最终会有多少阶段？

• 每个阶段用于并行加载两个数据源= 2个阶段
• 表示这join一点的第三阶段取决于其他两个阶段
• 注意:处理联接数据的所有后续操作可能在同一阶段执行,因为它们必须按顺序执行.启动其他阶段没有任何好处,因为他们无法在先前的操作完成之前开始工作.

这是玩具计划

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

现在:有多少任务？任务数量应该等于

(Stage*#Partitions in the stage)的总和

Answer 2

这可能有助于您更好地理解不同的部分:

• 阶段:是一组任务.针对不同数据子集(分区)运行的相同进程.
• 任务:表示分布式数据集分区上的工作单元.因此,在每个阶段,任务数量=分区数,或者您说"每个分区每个阶段一个任务".
• 每个执行器在一个纱线容器上运行,每个容器驻留在一个节点上.
• 每个阶段使用多个执行器,每个执行器分配多个vcores.
• 每个vcore一次只能执行一个任务
• 因此,在任何阶段,可以并行执行多个任务.正在运行的任务数=正在使用的正在使用的数量.

Answer 3

如果我理解正确,有2个(相关的)事情让你感到困惑:

1)什么决定了任务的内容？

2)什么决定了要执行的任务数量？

Spark的引擎在连续的rdds上"粘合" 简单的操作,例如:

rdd1 = sc.textFile( ... )
rdd2 = rdd1.filter( ... )
rdd3 = rdd2.map( ... )
rdd3RowCount = rdd3.count

因此当rdd3(懒惰地)计算时,spark将为rdd1的每个分区生成一个任务,并且每个任务将执行过滤器和每行的映射以产生rdd3.

任务数由分区数决定.每个RDD都有一个已定义数量的分区.对于从HDFS读取的源RDD(例如,使用sc.textFile(...)),分区数是由输入格式生成的分割数.RDD上的某些操作可能导致RDD具有不同数量的分区:

rdd2 = rdd1.repartition( 1000 ) will result in rdd2 having 1000 partitions ( regardless of how many partitions rdd1 had ).

另一个例子是加入:

rdd3 = rdd1.join( rdd2  , numPartitions = 1000 ) will result in rdd3 having 1000 partitions ( regardless of partitions number of rdd1 and rdd2 ).

(大多数)改变分区数量的操作涉及一个shuffle,当我们这样做时:

rdd2 = rdd1.repartition( 1000 ) 

实际发生的是rdd1的每个分区上的任务需要产生一个可以被下一阶段读取的结束输出,以便使rdd2有1000个分区(他们是如何做的？哈希或排序).这方面的任务有时被称为"地图(侧面)任务".稍后在rdd2上运行的任务将作用于一个分区(rdd2!),并且必须弄清楚如何读取/组合与该分区相关的地图侧输出.这方面的任务有时被称为"减少(侧面)任务".

这两个问题是相关的:一个阶段中的任务数量是分区的数量(连续的rdds"胶合"在一起)和rdd的分区数量可以在阶段之间改变(通过指定一些分区的数量) shuffle导致操作例如).

一旦阶段的执行开始,其任务可以占用任务时隙.并发任务槽的数量是numExecutors*ExecutorCores.通常,这些可以由来自不同的非依赖阶段的任务占据.