摸鱼大数据——Spark Core——Spark内核调度

1、内容概述

Spark内核调度的任务：

• 如何构建DAG执行流程图

• 如何划分Stage阶段

• Driver底层是如何运转

• 确定需要构建多少分区（线程）

Spark内核调度的目的：尽可能用最少的资源高效地完成任务计算

2、RDD的依赖

RDD依赖：一个RDD的形成可能是由一个或者多个RDD得到的，此时这个RDD和之前的RDD之间产生依赖关系。

在Spark中，RDD之间的依赖关系，主要有二种类型：

• 窄依赖：

作用: 能够让Spark程序并行计算。也就是一个分区数据计算出现问题以后，其他的分区计算不受到任何影响
特点: 父RDD的分区和子RDD的分区关系是一对一的关系。也就是父RDD分区的数据会整个被下游子RDD的分区接收

• 宽依赖：

作用: 划分Stage的重要依据。宽依赖也叫做Shuffle依赖
特点: 父RDD的分区和子RDD的分区关系是一对多的关系。也就是父RDD的分区数据会被分成多份给到下游子RDD的多个分区所接收。
​
注意: 如果有宽依赖，shuffle下游的其他操作，必须等待shuffle执行完成以后才能够继续执行。为了避免数据不完整

说明：

在实际使用中，不需要纠结哪些算子会存在shuffle，以需求为目标。虽然shuffle的存在会影响一定的效率, 但是以完成任务为准则，该用那个算子，就使用那个算子即可，不要过分纠结。

算子中一般以ByKey结尾的会发生shuffle；另外是重分区算子会发生shuffle

3、DAG和Stage

DAG：有向无环图，主要描述一段执行任务，从开始一直往下走，不允许出现回调操作

Spark应用程序中，遇到一个Action算子，就会触发形成一个Job任务的产生。

对于每一个Job的任务，都会产生一个DAG执行流程图，那么这个流程图是如何形成的呢?

层级关系：
1- 一个application应用程序 -> 遇到一个Action算子，就会触发形成一个Job任务
2- 一个Job任务只有一个DAG有向无环图
3- 一个DAG有向无环图 -> 有多个Stage
4- 一个Stage -> 有多个Task线程
​
5- 一个RDD -> 有多个分区
6- 一个分区会被一个Task线程所处理

DAG执行流程图形成和Stage划分：

1- Spark应用程序遇到Action算子后，就会触发一个Job任务的产生。Job任务会将它所依赖的所有算子全部加载进来，形成一个Stage
​
2- 接着从Action算子从后往前进行回溯，遇到窄依赖就将算子放在同一个Stage当中；如果遇到宽依赖，就划分形成新的Stage。最后一直回溯完成

细化剖析Stage内部的流程：

默认并行度的值确认:

因为是使用textFile读取HDFS上的文件，因此RDD分区数=max(文件的block块的数量, defaultMinPartition)。继续需要知道defaultMinPartition的值是多少。
​
defaultMinPartition=min(spark.default.parallelism,2)取最小值。最终我们确认spark.default.parallelism的参数值就能够最终确认RDD的分区数有多少个
​
spark.default.parallelism参数值确认过程如下：
1- 如果有父RDD，就取父RDD的最大分区数
2- 如果没有父RDD，根据集群模式进行取值：
   2.1- 本地模式：机器的最大CPU核数
   2.2- （了解）Mesos：默认是8
   2.3- 其他模式：所有执行节点上的核总数或2，以较大者为准

4、Spark Shuffle

Spark中shuffle的发展历程:

1- 在1.1版本以前，Spark采用Hash shuffle (优化前 和 优化后)
​
2- 在1.1版本的时候，Spark推出了Sort Shuffle
​
3- 在1.5版本的时候，Spark引入钨丝计划(优化为主)
​
4- 在1.6版本的时候，将钨丝计划合并到sortShuffle中
​
5- 在2.0版本的时候，将Hash Shuffle移除，将Hash shuffle方案移植到Sort Shuffle

未优化的Hash shuffle:

存在的问题：
    上游（map端）的每个Task会产生与下游Task个数相等的小文件个数。这种情况会导致上游有非常多的小文件。另外，下游（reduce端）来拉取文件的时候，会有大量的网络IO和磁盘IO过程，因为要打开和读取多个小文件。

经过优化后的Hash shuffle

变成了由每个Executor进程产生与下游Task个数相等的小文件数。这样可以大量减小小文件的产生，以及降低下游拉取文件时候的网络IO和磁盘IO过程

Sort shuffle:

Sort Shuffle分成了两种: 普通机制和bypass机制。具体使用哪种，由Spark底层决定。
​
普通机制的运行过程: 每个上游Task线程处理数据，数据处理完以后，先放在内存中。接着对内存中的数据进行分区、排序。将内存中的数据溢写到磁盘，形成一个个的小文件。溢写完成以后，会将多个小文件合并成一个大的磁盘文件。并且针对每个大的磁盘文件，会提供一个索引文件。接着是下游Task根据索引文件来读取相应的数据。
​
bypass机制: 就是在普通机制的基础上，省略了排序的过程
​
bypass机制的触发条件是：
1- 上游RDD的分区数量最多不能超过200个
2- 上游不能对数据进行提前聚合操作（因为提前聚合，需要先进行分组操作，而分组的操作实际上是有排序的操作）

5、Job调度流程

主要是讨论：在Driver内部，是如何调度任务

1- Driver进程启动后，底层PY4J创建SparkContext顶级对象。在创建该对象的过程中，还会创建另外两个对象，分别是: DAGScheduler和TaskScheduler
    DAGScheduler: DAG调度器。将Job任务形成DAG有向无环图和划分Stage的阶段
    TaskScheduler: Task调度器。将Task线程分配给到具体的Executor执行
​
2- 一个Spark程序遇到一个Action算子就会触发产生一个Job任务。SparkContext将Job任务给到DAG调度器，拿到Job任务后，会将Job任务形成DAG有向无环图和划分Stage的阶段。并且会确定每个Stage阶段有多少个Task线程，会将众多的Task线程放到TaskSet的集合中。DAG调度器将TaskSet集合给到Task调度器
​
3- Task调度器拿到TaskSet集合以后，将Task分配给到给到具体的Executor执行。底层是基于SchedulerBackend调度队列来实现的。
​
4- Executor开始执行任务。并且Driver会监控各个Executor的执行状态，直到所有的Executor执行完成，就认为任务运行结束
​
5- 后续过程和之前一样

6、Spark RDD并行度

整个Spark应用中，影响并行度的因素有以下两个原因:

• 1- 资源的并行度: Executor数量 和 CPU核心数 以及 内存的大小

• 2- 数据的并行度: Task的线程数 和 分区数量

一般将Task线程数设置为CPU核数的2-3倍。另外每个线程分配3-5GB的内存资源。

如何设置并行度:

语法: SparkConf().set("spark.default.parallelism", "num")
​
​
说明: spark.default.parallelism该参数是SparkCore中的参数。该参数只会影响shuffle以后的分区数量。另外该参数对parallelize并行化本地集合创建的RDD不起作用。

代码演示：

# 导包
from pyspark import SparkContext, SparkConf
import os

# 绑定指定的python解释器
os.environ['SPARK_HOME'] = '/export/server/spark'
os.environ['PYSPARK_PYTHON'] = '/root/anaconda3/bin/python3'
os.environ['PYSPARK_DRIVER_PYTHON'] = '/root/anaconda3/bin/python3'
if __name__ == '__main__':
    # 6.1 TODO: 设置并行度
    conf = SparkConf().set('spark.default.parallelism', '6')

    # - 1.创建SparkContext对象
    sc = SparkContext(conf=conf)
    # - 2.数据输入
    textRDD = sc.textFile('file:///export/data/spark_project/spark_base/content.txt')
    # - 3.数据处理
    #   - 3.1文本内容切分
    flatMapRDD = textRDD.flatMap(lambda line: line.split(" "))
    #   - 3.2数据格式转换
    mapRDD = flatMapRDD.map(lambda word: (word, 1))

    # 6.2 TODO: shuffle之前查看分区数量
    print(f"shullfe之前分区数: {mapRDD.getNumPartitions()},分区内容:{mapRDD.glom().collect()}")

    #   - 3.3分组和聚合
    reduceRDD = mapRDD.reduceByKey(lambda agg, curr: agg + curr)
    
    # 6.3 TODO: shuffle之后查看分区数量
    print(f"shullfe之后分区数: {reduceRDD.getNumPartitions()},分区内容:{reduceRDD.glom().collect()}")

    # - 4.数据输出
    # print(reduceRDD.collect())
    # - 5.释放资源
    sc.stop()