Spark内核(执行原理)之环境准备/Spark作业提交流程

太过爱你忘了你带给我的痛 2022-09-10 06:10 221阅读 0赞

了解Spark任务的提交到运行的流程，其中包含两个阶段：

第一阶段在Yarn集群之外执行，主要是作业的提交，将作业提交到Yarn集群为止。

第二阶段在Yarn集群执行，涉及资源分配、AM、Driver、Executor等组件的创建和启动。

![watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBAVFJYMTAyNA_size_20_color_FFFFFF_t_70_g_se_x_16][]

#  一、第一阶段：作业提交 #

1.提交脚本

作业提交脚本
    一般提交一个spark作业的方式采用spark-submit来提交
    # Run on a Spark standalone cluster
    ./bin/spark-submit \
      --class org.apache.spark.examples.SparkPi \
      --master spark://207.184.161.138:7077 \
      --executor-memory 20G \
      --total-executor-cores 100 \
      /path/to/examples.jar \
      1000

2.使用spark-submit 命令进行提交，看下命令脚本内容

#!/usr/bin/env bash
    # 、注释
    if [ -z "${SPARK_HOME}" ]; then
      source "$(dirname "$0")"/find-spark-home
    fi
    # disable randomized hash for string in Python 3.3+
    export PYTHONHASHSEED=0
    exec "${SPARK_HOME}"/bin/spark-class org.apache.spark.deploy.SparkSubmit "$@"

3.其实就是调用 org.apache.spark.deploy.SparkSubmit，即提交作业的起点

def doSubmit(args: Array[String]): Unit = {
        // Initialize logging if it hasn't been done yet. Keep track of whether logging needs to
        // be reset before the application starts.
        val uninitLog = initializeLogIfNecessary(true, silent = true)
    
        val appArgs = parseArguments(args)
        if (appArgs.verbose) {
          logInfo(appArgs.toString)
        }
        appArgs.action match {
          case SparkSubmitAction.SUBMIT => submit(appArgs, uninitLog)
          case SparkSubmitAction.KILL => kill(appArgs)
          case SparkSubmitAction.REQUEST_STATUS => requestStatus(appArgs)
          case SparkSubmitAction.PRINT_VERSION => printVersion()
        }
      }

做了两件事：

1.  解析命令参数
2.  调用submit()

4.submit()向yarn提交任务

// 通过设置适当的类路径，系统属性和应用程序参数来准备启动环境，以便基于集群管理和部署模式运行子主类。
        val (childArgs, childClasspath, sparkConf, childMainClass) = prepareSubmitEnvironment(args)
        方法中会为spark提交做准备，准备好运行环境相

*  准备启动环境
 *  YarnClusterApplication类创建client调用submitApplication()提交任务到yarn

5.现在任务已经提交到Yarn 的 ResourcesManager 第一阶段结束

# 二、第二阶段 Yarn集群 #

再把流程图贴一下：

![watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBAVFJYMTAyNA_size_20_color_FFFFFF_t_70_g_se_x_16 1][]

说明：

【RM】表示在RM上的操作

【AM】表示在AM上的操作

【Executor】表示在NM上的 Executor 进程的操作

步骤：

1.【RM】ResourcesManager 启动 ApplicationManager（AM）进程

client会先申请向RM一个Container，来启动AM进程

在ApplicationManager 中

*  有个 yarnRmClinet 用来跟RM进行通信
 *  有个nmClient 用来跟其他节点进行通信

2.【AM】AM根据参数启动Driver线程，并初始化SparkContext

3.【AM】通过yarnRmClinet与RM进行连接，注册AM，申请资源

4.【RM】返回资源可用列表

根据 首选位置 等相关参数信息分配容器资源

5.【AM】处理可分配的容器

此时已得到分配到的容器信息

运行已分配的容器（向指定的NM启动容器）

6.【AM】AM的线程池中在分配到的已运行的容器中启动 Executor 进程

7.【Executor】进入Executor

重要属性：

*  driver对象，与AM中的Driver进行通信
 *  ExecutorEnv Executor环境
 *  MassageLoop 收件箱发件箱

8.【Executor】driver对象向Driver注册Executor（反向注册）

9.【AM】Driver 回复 Executor 注册成功

10.【Executor】创建并启动 Executor 计算对象

# 三、**Spark** **通用运行流程** #

![watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBAVFJYMTAyNA_size_20_color_FFFFFF_t_70_g_se_x_16 2][]

上图为 Spark 通用运行流程图，体现了基本的 Spark 应用程序在部署中的基本提交流程。

这个流程是按照如下的核心步骤进行工作的：

1) 任务提交后，都会先启动 Driver 程序；

2) 随后 Driver 向集群管理器注册应用程序；

3) 之后集群管理器根据此任务的配置文件分配 Executor 并启动；

4) Driver 开始执行 main 函数，Spark 查询为懒执行，当执行到 Action 算子时开始反向推

算，根据宽依赖进行 Stage 的划分，随后每一个 Stage 对应一个 Taskset，Taskset 中有多

个 Task，查找可用资源 Executor 进行调度；

5) 根据本地化原则，Task 会被分发到指定的 Executor 去执行，在任务执行的过程中，

Executor 也会不断与 Driver 进行通信，报告任务运行情况。

[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBAVFJYMTAyNA_size_20_color_FFFFFF_t_70_g_se_x_16]: /images/20220829/9392d8427dc54f50af6c242c34f31740.png
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBAVFJYMTAyNA_size_20_color_FFFFFF_t_70_g_se_x_16 1]: /images/20220829/76f7c447f7844abb81e4b831695ccf06.png
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBAVFJYMTAyNA_size_20_color_FFFFFF_t_70_g_se_x_16 2]: /images/20220829/b24864c548454726874b515ca7b9a0ef.png