sparkjavajob_一篇文章搞清spark任务如何执行

本文将针对spark中的Driver和Executor讲起&＃xff0c;简述了spark的运行流程&＃xff0c;部署模式以及内部任务调度机制&＃xff0c;希望针对spark任务执行过程进行尽可能好理解的解析

1.两个重要的主角

​在spark中&＃xff0c;有两个重要的主角是绕不开的&＃xff0c;driver和executor&＃xff0c;他们的结构呈一主多从模式&＃xff0c;driver就是那个单身狗&＃xff0c;控制欲很强&＃xff0c;权利也很大&＃xff0c;每天独自一人没别的事&＃xff0c;就想法设法的指挥着手下一堆executor到处干活。他们分工明确&＃xff0c;组织结构简单&＃xff0c;共同支撑起了spark强大的计算引擎。

Driver

​Spark 驱动器节点&＃xff0c;用于执行 Spark 任务中的 main 方法&＃xff0c;负责实际代码的执行工作。Driver 在 Spark 作业执行时主要负责&＃xff1a;

1. 将代码逻辑转化为任务&＃xff1b;

2. 在 Executor 之间调度任务(job)&＃xff1b;

3. 跟踪 Executor 的执行情况(task)。

Executor

​Spark 执行器节点&＃xff0c;负责在 Spark 作业中运行具体任务&＃xff0c;任务之间相互独立。Spark 应用启动时&＃xff0c;Executor 节点被同时启动&＃xff0c;并且始终伴随着整个 Spark 应用的生命周期而存在。如果有 Executor 点发生了故障或崩溃&＃xff0c;Spark 应用也可以继续执行&＃xff0c;会将出错节点上的任务调度到其他 Executor 节点上继续运行。Executor 有两个核心功能&＃xff1a;

1. 负责运行组成 Spark 应用的任务&＃xff0c;并将结果返回给驱动器进程&＃xff1b;

2. 通过自身的块管理器(Block Manager)为用户程序中要求缓存的 RDD提供内存式存储。RDD 是直接缓存在 Executor 进程内的&＃xff0c;因此任务可以在运行时充分利用缓存数据加速运算。

Spark 运行流程

​不论spark以何种方式部署&＃xff0c;在任务提交后&＃xff0c;都先启动Driver&＃xff0c;然后Driver向集群管理器注册应用程序&＃xff0c;之后集群管理器根据此任务的配置文件分配Executor并启动&＃xff0c;然后Driver等待资源满足&＃xff0c;执行 main 函数&＃xff0c;Spark的查询为懒执行&＃xff0c;当执行到 action 算子时才开始真正执行&＃xff0c;开始反向推算&＃xff0c;根据宽依赖进行 stage 的划分&＃xff0c;随后每一个 stage 对应一个 taskset&＃xff0c;一个taskset 中有多个 task&＃xff0c;task 会被分发到指定的 Executor 去执行&＃xff0c;在任务执行的过程中&＃xff0c;Executor 也会不断与 Driver 进行通信&＃xff0c;报告任务运行情况。

2.spark的部署模式

2.1 spark部署类型

Spark共支持3种集群管理器&＃xff0c;Standalone&＃xff0c;Mesos和Yarn

Standalone&＃xff1a;

独立模式&＃xff0c;Spark 原生的最简单的一个集群管理器。 它可以运行在各种操作系统上&＃xff0c;自带完整的服务&＃xff0c;无需依赖任何其他资源管理系统&＃xff0c;使用 Standalone 可以很方便地搭建一个集群。

Apache Mesos

Mesos也是一个强大的分布式资源管理框架&＃xff0c;是以与Linux内核同样的原则而创建的&＃xff0c;允许多种不同的框架部署在其上

Hadoop Yarn

Hadoop生态下的统一资源管理机制&＃xff0c;在上面可以运行多套计算框架&＃xff0c;如mapreduce、spark 等&＃xff0c;根据 driver 在集群中的位置不同&＃xff0c;部署模式可以分为 yarn-client 和 yarn-cluster。

Spark 的运行模式取决于传递给 SparkContext 的 MASTER 环境变量的值&＃xff0c;spark在yarn上部署:

yarn-client&＃xff1a;Driver在本地&＃xff0c;Executor在Yarn集群&＃xff0c;配置&＃xff1a;--deploy-mode client

yarn-cluster&＃xff1a;Driver和Executor都在Yarn集群&＃xff0c;配置&＃xff1a;--deploy-mode cluster

2.2 Yarn模式下的运行机制

当前流行的工作模式均是将spark提交到Yarn上&＃xff0c;所以这里我们针对spark on Yarn做一下详细了解。

yarn-client 模式

​ 在YARNClient模式下&＃xff0c;Driver在任务提交的本地机器上运行&＃xff0c;Driver会向ResourceManager申请启动ApplicationMaster&＃xff0c;随后ResourceManager分配container&＃xff0c;在合适的NodeManager上启动ApplicationMaster&＃xff0c;此时的ApplicationMaster的功能相当于一个ExecutorLaucher&＃xff0c;只负责向ResourceManager申请Executor内存。

ResourceManager接到ApplicationMaster的资源申请后会分配container&＃xff0c;然后

ApplicationMaster在资源分配指定的NodeManager上启动Executor进程&＃xff0c;Executor进程启动后会向Driver反向注册。另外一条线&＃xff0c;Driver自身资源满足的情况下&＃xff0c;Driver开始执行main函数&＃xff0c;之后执行Action算子时&＃xff0c;触发一个job&＃xff0c;并根据宽依赖开始划分stage&＃xff0c;每个stage生成对应的taskSet&＃xff0c;Executor注册完成后&＃xff0c;Driver将task分发到各个Executor上执行。

yarn-cluster

​ 在 YARN Cluster 模式下&＃xff0c;任务提交后会和 ResourceManager 通讯申请启动ApplicationMaster&＃xff0c;随后 ResourceManager 分配 container&＃xff0c;在合适的 NodeManager上启动 ApplicationMaster&＃xff0c;此时的ApplicationMaster 就是 Driver。

​ Driver 启动后向 ResourceManager 申请 Executor 内存&＃xff0c;ResourceManager会分配container&＃xff0c;然后在合适的 NodeManager 上启动 Executor 进程&＃xff0c;Executor 进程启动后会向 Driver 反向注册。另外一条线&＃xff0c;Driver自身资源满足的情况下&＃xff0c;开始执行main函数&＃xff0c;之后执行Action算子时&＃xff0c;触发一个job&＃xff0c;并根据宽依赖开始划分stage&＃xff0c;每个stage生成对应的taskSet&＃xff0c;Executor注册完成后&＃xff0c;Driver将task分发到各个Executor上执行。

3.Spark 任务调度

​Driver会根据用户程序准备任务&＃xff0c;并向Executor分发任务&＃xff0c;在这儿有几个Spark的概念需要先介绍一下&＃xff1a;

Job&＃xff1a;以Action算子为界&＃xff0c;遇到一个Action方法就触发一个Job

Stage&＃xff1a;Job的子集&＃xff0c;一个job至少有一个stage&＃xff0c;以shuffle(即RDD宽依赖)为界&＃xff0c;一个shuffle划分一个stage

Task&＃xff1a; Stage 的子集&＃xff0c;以并行度(分区数)来衡量&＃xff0c;分区数是多少&＃xff0c;则有多少

个 task。

spark在具体任务的调度中&＃xff0c;总的分两路进行&＃xff1a;Stage级别调度和Task级别调度。Spark RDD通过转换(Transactions)算子&＃xff0c;形成了血缘关系图DAG&＃xff0c;最后通过行动(Action)算子&＃xff0c;触发Job并调度执行。

DAGScheduler负责Stage级的调度&＃xff0c;主要是将DAG切分成若干Stages&＃xff0c;并将每个Stage打包成TaskSet交给TaskScheduler调度。

TaskScheduler负责Task级的调度&＃xff0c;将DAGScheduler给过来的TaskSet按照指定的调度策略分发到Executor上执行

3.1 Spark Stage级调度

​Spark的任务调度是从DAG切割开始&＃xff0c;主要是由DAGScheduler来完成。当遇到一个Action操作后就会触发一个Job的计算&＃xff0c;并交给DAGScheduler来处理。

DAGScheduler主要做两个部分的事情&＃xff1a;

切分stage

​DAGScheduler会根据RDD的血缘关系构成的DAG进行切分&＃xff0c;将一个Job划分为若干Stages&＃xff0c;具体划分策略是&＃xff1a;从后往前&＃xff0c;由最终的RDD不断通过依赖回溯判断父依赖是否是宽依赖&＃xff0c;遇到一个shuffle就划分一个Stage。无shuffle的称为窄依赖&＃xff0c;窄依赖之间的RDD被划分到同一个Stage中。划分的Stages分两类&＃xff0c;一类叫做ResultStage&＃xff0c;为DAG最下游的Stage&＃xff0c;由Action方法决定&＃xff0c;另一类叫做ShuffleMapStage&＃xff0c;为下游Stage准备数据。

​stage任务调度本身是一个反向的深度遍历算法&＃xff0c;以下图wordcount为例。此处只有saveAsTextFile为行动算子&＃xff0c;该 Job 由 RDD-3 和 saveAsTextFile方法组成&＃xff0c;根据依赖关系回溯&＃xff0c;知道回溯至没有依赖的RDD-0。回溯过程中&＃xff0c;RDD-2和RDD-3存在reduceByKey的shuffle&＃xff0c;会划分stage&＃xff0c;由于RDD-3在最后一个stage&＃xff0c;即划为ResultStage&＃xff0c;RDD-2&＃xff0c;RDD-1&＃xff0c;RDD-0&＃xff0c;这些依赖之间的转换算子flatMap&＃xff0c;map没有shuffle&＃xff0c;因为他们之间是窄依赖&＃xff0c;划分为ShuffleMapStage。

打包Taskset提交Stage

​ 一个Stage如果没有父Stage&＃xff0c;那么从该Stage开始提交&＃xff0c;父Stage执行完毕才能提交子Stage。Stage提交时会将Task信息(分区信息以及方法等)序列化并被打包成TaskSet交给TaskScheduler&＃xff0c;一个Partition对应一个Task&＃xff0c;另一方面TaskScheduler会监控Stage的运行状态&＃xff0c;只有Executor丢失或者Task由于Fetch失败才需要重新提交失败的Stage以调度运行失败的任务&＃xff0c;其他类型的Task失败会在TaskScheduler的调度过程中重试。

3.2 Spark Task 级调度

​SparkTask的调度是由TaskScheduler来完成&＃xff0c;TaskScheduler将接收的TaskSet封装为TaskSetManager加入到调度队列中。同一时间可能存在多个TaskSetManager&＃xff0c;一个TaskSetManager对应一个TaskSet&＃xff0c;而一个TaskSet含有n多个task信息&＃xff0c;这些task都是同一个stage的。

​TaskScheduler初始化后会启动SchedulerBackend&＃xff0c;它负责跟外界打交道&＃xff0c;接收Executor的注册信息&＃xff0c;并维护Executor的状态&＃xff0c;SchedulerBackend会监控到有资源后&＃xff0c;会询问TaskScheduler有没有任务要运行&＃xff0c;TaskScheduler会从调度队列中按照指定的调度策略选择TaskSetManager去调度运行。

​TaskSetManager按照一定的调度规则一个个取出task给TaskScheduler&＃xff0c;TaskScheduler再交给SchedulerBackend去发到Executor上执行。

​Task被提交到Executor启动执行后&＃xff0c;Executor会将执行状态上报给SchedulerBackend&＃xff0c;SchedulerBackend则告诉TaskScheduler&＃xff0c;TaskScheduler找到该Task对应的TaskSetManager&＃xff0c;并通知到该TaskSetManager&＃xff0c;这样TaskSetManager就知道Task的运行状态

3.3 失败重试和白名单

​对于运行失败的Task&＃xff0c;TaskSetManager会记录它失败的次数&＃xff0c;如果失败次数还没有超过最大重试次数&＃xff0c;那么就把它放回待调度的Task池子中等待重新执行&＃xff0c;当重试次数过允许的最大次数&＃xff0c;整个Application失败。在记录Task失败次数过程中&＃xff0c;TaskSetManager还会记录它上一次失败所在的ExecutorId和Host&＃xff0c;这样下次再调度这个Task时&＃xff0c;会使用黑名单机制&＃xff0c;避免它被调度到上一次失败的节点上&＃xff0c;起到一定的容错作用。

关于找一找教程网

本站文章仅代表作者观点&＃xff0c;不代表本站立场&＃xff0c;所有文章非营利性免费分享。

本站提供了软件编程、网站开发技术、服务器运维、人工智能等等IT技术文章&＃xff0c;希望广大程序员努力学习&＃xff0c;让我们用科技改变世界。

[一篇文章搞清spark任务如何执行]http://www.zyiz.net/tech/detail-120141.html