Spark 内核机制(重要)

image.png

Spark 内核机制 就是Driver中 Job怎样调度task任务的过程

提交流程:

1. submit_client 提交应用给rm.rm 再响应的nodemanager上开辟容器,运行am (driver)
   2,am再想tm 申请资源 运行executor, rm根据提交的内容分配容器启动executor 执行者
   3.每个executor启动会跟AM反向注册
   4.当executor 全部跟 Driver 反向注册之后,Driver 开始从main 函数进行读取,当读到一个触发算子时,采用回溯法,构建DAG调度一个job内有很多stage阶段,(当读到一个宽依赖的算子时 划分stage),每个stage内有很多task 任务(根据窄依赖算子),task的个数取决于本阶段最后一个RDD的分区个数.

注意 每个stage内的 spark 计算是pipeline 管道计算,即根据最后一个rdd的分区的分区个数,划分task任务的个数,建立一系列的管道 ,然后跟其他窄依赖的rdd的分区简历管道,一个数据 一个数据的处理

image.png

内核调度流程

image.png

1.当所有的executor 向Driver 注册完成后开始读取main方法中的代码
自上而下执行
2..首先对sparkcontext对象进行初始化,当读到一action算子的时候由DAGScheduler采用回溯法构建DAG调度图,依据RDD中的宽依赖划分stage(会产生shuffle)
3.每个stage中又由 taskScheduler 根据最后阶段内最后一个rdd的分区数目确定task任务的个数,每个task的计算是管道计算(pipeline)最后调度stage中所有的task任务到executor上去执行

每个task处理完数据 写入磁盘当中 shufflewriter 有三种方式实现
shuffle中分区规则(hash分区,range(范围)分区)

每个stage中task计算模式:pipeline 管道计算模式
即先建立每个RDD直接的管道,然后一个数据一个数据的计算

RDD依赖关系

• 窄依赖
  • 父RDD的一个分区的数据 只给子RDD的1个分区 (一对一的关系)
  • 不产生shuffle 如果子RDD的某个分区数据丢失，重构父RDD的对应分区

注意 分区调整算子coalesce和`repartition 也是属于窄依赖 同一个stage
map算子 filter算子等

• 宽依赖(shuffle 依赖 )
  • 父RDD的一个分区的数据 只给子RDD的多个分区(一对多的关系)
  • 会产生shuffle,如果子RDD的某个分区数据丢失，必须重构父RDD的所有分区

groupbykey reducebykey

spark shuffle

image.png
shufflewrite分为三种

• 第1种：SortShuffleWriter，普通机制

  • 第1步、先将数据写入内存，达到一定大小，进行排序

  • 第2步、再次写入内存缓冲，最后写入磁盘文件

  • 第3步、最终合并一个文件和生成索引文件

• 第2种：BypassMergeSortShuffleWriter，bypass机制

  • 当map端不用聚合，并且partition分区数目小于200时，采用该机制

  • 第1步、直接将数据写入内存缓冲，再写入磁盘文件

  • 第2步、最后合并一个文件和生成索引文件

• 第3种：UnsafeShuffleWriter，钨丝优化机制

  • 当map端不用聚合，分区数目小于16777215，并且支持relocation序列化

  • 第1步、利用Tungsten的内存作为缓存，将数据写入到缓存，达到一定大小写入磁盘

  • 第2步、最后合并一个文件和生成索引文件

具体流程

当启动Spark Application的时候，运行MAIN函数，首先创建SparkContext对象，此时构建DAG调度器DAGScheduler和Task任务调度器TaskScheduler实例对象。

1. DAGScheduler实例对象
   • 将每个Job的DAG图划分为Stage，依据RDD之间依赖为宽依赖（产生Shuffle）
2. TaskScheduler实例对象

• 调度每个Stage中所有Task：TaskSet，发送到Executor上执行
• 每个Stage中会有多个Task，所有Task处理数据不一样（每个分区数据被1个Task处理），但是处理逻辑一样的。

• 将每个Stage中所有Task任务，放在一起称为TaskSet。

  
  image.png