spark基本架构与运行原理

spark运行架构图

• Cluster Manager：在standalone模式中即为Master主节点，控制整个集群，监控worker。在YARN模式中为ResourceManager
• Worker:从节点，负责控制计算节点，启动executor。在yarn集群中为nodemanager

• Driver:运行Application的main函数并且创建SparkContext.
• Executor:执行器,Application运行在worker节点上的一个进程，负责启动线程池运行任务(Task)
• RDD DAG:RDD组成的有向无环图，反应RDD之间的依赖关系。当RDD遇到遇到Action算子的时候，将之前所有算子形成一个有向无环图
• DAG Scheduler：根据job构造基于stage的DAG并将stage提交给TaskScheduler
• TaskScheduler:将task任务分发给Executor执行
• SparkEnv:线程级别的上下文，存储运行时中要组件的引用

Spark作业基本概念

• Application:用户编写的Spark应用程序，其中包括一个Driver功能的代码和分布在集群中多个节点上运行的Executor代码
• Driver Program:运行Application的main函数并且创建SparkContext
• Job:一个RDD Graph 触发的作业，包含多个Task组成的并行计算
• Stage:每个job会根据RDD的宽依赖关系切分成多个stage,每个stage包含一组相同的Task,也叫TaskSet
• Task：被送到某个Executor上的工作单元，但hadoopMR中的MapTask和ReduceTask概念一样，是运行Application的基本单位，多个Task组成一个Stage，而Task的调度和管理等是由TaskScheduler负责

Spark运行机制

spark运行流程

1. 用户在客户端将Application提交到ClusterManager
2. ClusterManager收到Application后，找一个worker启动Driver,初始化一个SparkContext.
3. SparkContext向资源管理器申请Executors,并启动StandaloneExecutorBackend
4. Executors向SparkContext申请Task,SparkContext中的DAG Scheduler会根据依赖关系拆分成多个stage,每个stage提交给一个Task Scheduler，Task Scheduler会将这些Task任务分发到Executor
5. Executor启动多线程，执行Task。
6. 所有Task完成后，SparkContext向Master注销

Spark运行模式

• Standalone:Standalone模式使用Spark自带的资源调度框架,采用Master/Slaves的典型架构

1. 用户在客户端将Application提交到ClusterManager
2. ClusterManager收到Application后，找一个worker启动Driver,初始化一个SparkContext.
3. SparkContext初始化时创建DAG Scheduler和Task Scheduler,向资源管理器申请Executors,并启动StandaloneExecutorBackend
4. Executors向SparkContext申请Task,SparkContext中的每一个DAG Scheduler会根据依赖关系拆分成多个stage,每个stage提交给一个Task Scheduler，Task Scheduler会将这些Task任务分发到Executor
5. Executor启动多线程，执行Task。
6. 所有Task完成后，SparkContext向Master注销

• Yarn Client:

1. Driver在本地提交任务的机器运行,Driver启动后向ResourceManager申请ApplicationMaster
2. ResourceManager找到一个NodeManager分配conainer,并且创建ApplicationMaster
3. AM创建后,向ResourceManager申请启动Executor,Executor启动后向Driver注册.
4. Driver注册完成后,Driver开始执行main方法,执行算子时,执行一个job,每个job会根据宽依赖划分为多个stage,每个stage,每个stage对应多个TaskSet,经过TaskSchedule分发到Executor执行.

• Yarn Cluster

1. 任务提交后,向ResourceManager申请ApplcationMaster
2. ResourceManager找到一个NodeManager分配conainer,并且创建ApplicationMaster,AM创建后,启动Driver,此时的AM就是一个Driver
3. AM向ResourceManager申请启动Executor,Executor启动后向Driver注册
4. Driver注册完成后,Driver开始执行main方法,执行算子时,执行一个job,每个job会根据宽依赖划分为多个stage,每个stage,每个stage对应多个TaskSet,经过TaskSchedule分发到Executor执行.

Spark Shuffle阶段

Spark Shuffle分为Map和Reduce阶段,它与stage阶段的划分有关,在一个job中如果碰到宽依赖,会被划分为多个stage,最终每个stage都会被划分为ResultStage和ShuffleMapStage,对应ReduceTask和ShuffleMapTask.ShuffleMapTask会根据partitioner对数据进行分组,并且进行持久化.

• 未优化HashShuffle

  HashShuffle在没有优化之前,每一个ShuffleMapTask会为每一个ReduceTask创建一个bucket(经过partitioner后对用的bucket),并且为每一个bucket创建一个文件.ReduceTask会去对应bucket中获取数据.但是这样有很大缺点就是生成的文件数量太多,这对于磁盘IO影响很大

• 优化后的HashShuffle:

  设置参数 spark.shuffle.consolidateFiles=true,在中间过程生成的文件数为cpu核数*reduceTask个数

• Sorted-Based Shuffle:

  1. SortShuffleWriter运行机制:类似hadoop的shuffle过程,在ShuffleMapTask阶段,每个Task将所有数据写到一个文件中,并且生成一个索引文件.然后在由ReduceTask去文件中拉取对应分区的数据.

  2. bypass运行机制:用于处理数据量小不需要排序和聚合的Shuffle操作

     触发条件:1. shuffle map task的数量小于200(spark.shuffle.sort.bypassMergeThreshold )

     ​ 2. 不是聚合类的shuffle算子