Spark底层原理：案例解析（第34天）

系列文章目录

一、spark架构设计概述
二、spark核心组件
三、spark架构设计举例分析
四、job调度流程详解
五、spark交互流程详解

前言

apache spark是一个快速、通用、基于内存的分布式计算引擎，专为大规模数据处理而设计。其架构设计体现了高度的模块化和可扩展性，支持多种数据处理模式，包括批处理、实时流处理、交互式查询、机器学习和图计算等。以下将详细介绍spark的架构设计，并结合具体例子进行分析。

一、spark架构设计概述

spark的架构设计遵循主从（master-slave）架构模式，主要由以下几部分组成：

1. 集群资源管理器（cluster manager）

• 负责集群资源的分配和管理，包括cpu、内存等资源。根据不同的部署模式，cluster manager可以是spark自带的standalone模式，也可以是yarn、mesos等第三方资源管理器。

2. 工作节点（worker node）

• 执行提交的任务，通过注册机制向cluster manager汇报自身的资源使用情况。在master的指示下，worker node会创建并启动executor进程，用于执行具体的计算任务。

3. 驱动程序（driver program/driver）

• 运行应用程序的main()函数，负责创建sparkcontext对象，并与cluster manager和executor进行通信，以协调任务的执行。
• 执行器（executor）：运行在worker node上的进程，负责执行driver分配的任务，并将结果返回给driver。executor是spark中真正的计算单元，它负责task的运行并将结果数据保存到内存或磁盘上。

二、spark核心组件

spark基于spark core建立了多个核心组件，每个组件都提供了特定的数据处理能力：

1. spark core

• 基础设施：包括sparkconf（配置信息）、sparkcontext（spark上下文）、spark rpc（远程过程调用）、listenerbus（事件总线）、metricssystem（度量系统）、sparkenv（环境变量）等，为spark的各种组件提供基础支持。
• 存储系统：spark的存储系统优先考虑在内存中存储数据，当内存不足时才会将数据写入磁盘。这种内存优先的存储策略使得spark在处理大规模数据时具有极高的性能。
• 调度系统：由dagscheduler和taskscheduler组成，负责任务的调度和执行。dagscheduler负责将用户程序转换为dag图，并根据依赖关系划分stage和task；taskscheduler则负责按照调度算法对task进行批量调度。
• 计算引擎：由内存管理器、任务管理器、task shuffle管理器等组成，负责具体的计算任务执行。

2. spark sql

• 提供基于sql的数据处理方式，支持结构化数据的查询和分析。spark sql可以将结构化数据（如json、csv、parquet等）转换为rdd或dataframe，并支持使用hive元数据和sql查询。

3. spark streaming

• 提供流处理能力，支持从kafka、flume、kinesis、tcp等多种数据源实时获取数据流，并将其转换为可供分析和存储的批处理数据。spark streaming使用dstream（离散流）作为数据流的抽象，并支持一系列的转换操作。

4. spark mllib

• 提供机器学习库，包括统计、分类、回归、聚类等多种机器学习算法的实现。spark mllib的分布式计算能力使得在大规模数据上进行机器学习任务成为可能。

5. spark graphx

• 提供图计算库，支持对大规模图结构数据进行处理和分析。graphx通过pregel提供的api可以快速解决图计算中的常见问题，如社交网络分析、网络拓扑分析等。

三、spark架构设计举例分析

• 以spark standalone模式为例，我们可以详细分析spark的架构设计如何支持数据处理任务的执行：

1. 集群启动：

• 在standalone模式下，集群由一个主节点（master）和多个工作节点（worker）组成。主节点负责管理集群资源并分配任务给工作节点；工作节点则负责执行具体的任务。
• 集群启动时，master节点会启动并监听来自worker节点的注册请求。worker节点在启动时向master注册，并报告自身的资源情况（如cpu、内存等）。

2. 任务提交：

• 用户通过driver程序提交spark作业到集群。driver程序首先创建sparkcontext对象，并连接到master节点以请求资源。
• master节点根据集群的资源情况和作业的资源需求，为driver分配资源，并启动executor进程。executor进程是运行在worker节点上的，用于执行具体的计算任务。

3. 任务执行：

• driver程序将作业划分为多个task，并通过executor的rpc接口将task发送到executor上执行。
• executor接收到task后，会在本地启动线程来并行执行task。执行过程中，executor会从存储系统（如hdfs）中加载数据，进行计算，并将结果返回给driver。
  driver收集所有executor的执行结果，并进行汇总和处理，最终将结果输出给用户。

4. 容错与恢复：

• spark通过rdd的容错机制来保证数据的可靠性和作业的可恢复性。rdd具有可容错性，当某个节点发生故障

四、job调度流程详解

• 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- 后续过程和之前一样

五、spark交互流程详解

1、client_spark集群

driver任务: driver进程中负责资源申请的工作并且负责创建sparkcontext对象的代码映射为java对象,进行创建任务的分配、任务的管理工作。

1- submit提交任务到主节点master

2- 在提交任务的那个客户端上启动driver进程

3- driver进程启动后，执行main函数,首先创建sparkcontext对象。底层是基于py4j，将创建sparkcontext对象的代码映射为java进行创建

4- driver进程连接到spark集群中的master主节点，根据资源配置要求，向主节点申请资源，用来启动executor

5- 主节点接收到资源申请之后，进行资源分配，底层是基于fifo（先进先出）。分配好资源资源之后，将方案返回给到driver进程
executor1：node1 2g 2cpu
executor2：node3 2g 2cpu

6-driver连接到对应的worker从节点上，占用相应的资源。通知worker启动executor进程。启动以后会反向注册回driver

7-driver开始处理代码
7.1- driver加载rdd相关的算子，根据算子间的依赖关系绘制dag有向无环图和划分stage阶段，并且确定每个stage阶段有多少个task线程。需要分配给哪些executor进行执行。
7.2- driver通知对应的executor进程来执行相应的任务
7.3- executor开始执行具体的任务。但是发现代码中有大量的python函数，而executor是jvm进程，无法直接执行代码。因此会调用服务器上的python解释器，将python函数和输入数据传输给到python解释器，执行完以后，将结果数据返回给executor进程
7.4- executor在运行过程中，会判断是否需要将结果数据返回给到driver进程。如果需要，就返回给driver进程；如果不需要，直接输出，结束即可。
7.5- driver会定时检查多个executor的执行状态。直到所有的executor执行完成，就认为任务运行结束

8- driver调用sc.stop()代码，通知master回收资源。整个程序运行结束。

2、cluster_spark集群

driver任务: driver进程中负责资源申请的工作并且负责创建sparkcontext对象的代码映射为java对象,进行创建任务的分配、任务的管理工作。

和client on spark集群的区别点: driver进程不是运行在提交任务的那台机器上了，而是在spark集群中随机选择一个worker从节点来启动和运行driver进程

1- submit提交任务到主节点master

2- master主节点接收到任务信息以后，根据driver的资源配置要求，在集群中随机选择（在资源充沛的众多从节点中随机选择）一个worker从节点来启动和运行driver进程

3- driver进程启动以后，执行main函数，首先创建sparkcontext对象。底层是基于py4j，将创建sparkcontext对象的代码映射为java进行创建

4- driver进程连接到spark集群中的master主节点，根据资源配置要求，向主节点申请资源，用来启动executor

5- 主节点接收到资源申请之后，进行资源分配，底层是基于fifo（先进先出）。分配好资源资源之后，将方案返回给到driver进程
executor1：node1 2g 2cpu
executor2：node3 2g 2cpu

6-driver连接到对应的worker从节点上，占用相应的资源。通知worker启动executor进程。启动以后会反向注册回driver

7-driver开始处理代码
7.1- driver加载rdd相关的算子，根据算子间的依赖关系绘制dag有向无环图和划分stage阶段，并且确定每个stage阶段有多少个task线程。需要分配给哪些executor进行执行。
7.2- driver通知对应的executor进程来执行相应的任务
7.3- executor开始执行具体的任务。但是发现代码中有大量的python函数，而executor是jvm进程，无法直接执行代码。因此会调用服务器上的python解释器，将python函数和输入数据传输给到python解释器，执行完以后，将结果数据返回给executor进程
7.4- executor在运行过程中，会判断是否需要将结果数据返回给到driver进程。如果需要，就返回给driver进程；如果不需要，直接输出，结束即可。
7.5- driver会定时检查多个executor的执行状态。直到所有的executor执行完成，就认为任务运行结束

8- driver调用sc.stop()代码，通知master回收资源。整个程序运行结束。

3、client on yarn集群

区别点: 将driver进程中负责资源申请的工作，转交给到yarn的applicationmaster来负责。driver负责创建sparkcontext对象的代码映射为java对象,进行创建任务的分配、任务的管理工作。

1- 首先会在提交的节点启动一个driver进程

2- driver进程启动以后，执行main函数，首先创建sparkcontext对象。底层是基于py4j，将创建sparkcontext对象的代码映射为java进行创建

3- 连接yarn集群的主节点（resourcemanager），将需要申请的资源封装为一个任务，提交给到yarn的主节点。主节点收到任务以后，首先随机选择一个从节点（nodemanager）启动applicationmaster

4- 当applicationmaster启动之后，会和yarn的主节点建立心跳机制，告知已经启动成功。启动成功以后，就进行资源的申请工作，将需要申请的资源通过心跳包的形式发送给到主节点。主节点接收到资源申请后，开始进行资源分配工作，底层是基于资源调度器来实现（默认为capacity容量调度器）。当主节点将资源分配完成以后，等待applicationmaster来拉取资源。applicationmaster会定时的通过心跳的方式询问主节点是否已经准备好了资源。一旦发现准备好了，就会立即拉取对应的资源信息。

5- applicationmaster根据拉取到的资源信息，连接到对应的从节点。占用相应的资源，通知从节点启动executor进程。从节点启动完executor之后，会反向注册回driver进程

6-driver开始处理代码
6.1- driver加载rdd相关的算子，根据算子间的依赖关系绘制dag有向无环图和划分stage阶段，并且确定每个stage阶段有多少个task线程。需要分配给哪些executor进行执行。
6.2- driver通知对应的executor进程来执行相应的任务
6.3- executor开始执行具体的任务。但是发现代码中有大量的python函数，而executor是jvm进程，无法直接执行代码。因此会调用服务器上的python解释器，将python函数和输入数据传输给到python解释器，执行完以后，将结果数据返回给executor进程
6.4- executor在运行过程中，会判断是否需要将结果数据返回给到driver进程。如果需要，就返回给driver进程；如果不需要，直接输出，结束即可。
6.5- driver会定时检查多个executor的执行状态。直到所有的executor执行完成，就认为任务运行结束。同时applicationmaster也会接收到各个节点的执行完成状态，然后通知主节点。任务执行完成了，主节点回收资源，关闭applicationmaster，并且通知driver。

7- driver执行sc.stop()代码。driver进程退出

4、cluster on yarn集群

区别点: 在集群模式下，driver进程的功能和applicationmaster的功能（角色）合二为一了。driver就是applicationmaster，applicationmaster就是driver。既要负责资源申请，又要负责任务的分配和管理。

1- 首先会将任务提交给yarn集群的主节点（resourcemanager）

2- resourcemanager接收到任务信息后，根据driver(applicationmaster)的资源配置信息要求，选择一个
nodemanager节点(有资源的，如果都有随机)来启动driver(applicationmaster)程序，并且占用相对应资源

3- driver(applicationmaster)启动后，执行main函数。首先创建sparkcontext对象（底层是基于py4j，识
别python的构建方式，将其映射为java代码）。创建成功后，会向resourcemanager进行建立心跳机制，告知已经
启动成功了

4- 根据executor的资源配置要求，向resourcemanager通过心跳的方式申请资源，用于启动executor（提交的任
务的时候，可以自定义资源信息）

5- resourcemanager接收到资源申请后，根据申请要求，进行分配资源。底层是基于资源调度器来资源分配（默认
为capacity容量调度）。然后将分配好的资源准备好，等待driver(applicationmaster)拉取操作
executor1： node1 2个cpu 2gb内存
executor2： node3 2个cpu 2gb内存

6- driver(applicationmaster)会定时询问是否准备好资源，一旦准备好，立即获取。根据资源信息连接对应的
节点，通知nodemanager启动executor，并占用相应资源。nodemanager对应的executor启动完成后，反向注册
回给driver(applicationmaster)程序（已经启动完成）

7- driver(applicationmaster)开始处理代码：
7.1 首先会加载所有的rdd相关的api(算子)，基于算子之间的依赖关系，形成dag执行流程图，划分stage阶
段，并且确定每个阶段应该运行多少个线程以及每个线程应该交给哪个executor来运行（任务分配）
7.2 driver(applicationmaster)程序通知对应的executor程序， 来执行具体的任务
7.3 executor接收到任务信息后， 启动线程， 开始执行处理即可： executor在执行的时候， 由于rdd代
码中有大量的python的函数，executor是一个jvm程序 ，无法解析python函数， 此时会调用python解析器，执
行函数， 并将函数结果返回给executor
7.4 executor在运行过程中，如果发现最终的结果需要返回给driver(applicationmaster)，直接返回
driver(applicationmaster)，如果不需要返回，直接输出 结束即可
7.5 driver(applicationmaster)程序监听这个executor执行的状态信息，当executor都执行完成后，
driver(applicationmaster)认为任务运行完成了

8- 当任务执行完成后，driver执行sc.stop()通知resourcemanager执行完成，resourcemanager回收资源，
driver程序退出即可