Spark系列之Spark体系架构

Posted 2023-03-30 落叶飘雪2014

---

title: Spark系列

---

第四章 Spark体系架构

4.1 Spark核心功能

Alluxio  原来叫   tachyon    分布式内存文件系统

Spark Core提供Spark最基础的最核心的功能，主要包括：

1、SparkContext

通常而言，DriverApplication的执行与输出都是通过SparkContext来完成的，在正式提交Application之前，首先需要初始化SparkContext。

SparkContext隐藏了网络通信、分布式部署、消息通信、存储能力、计算能力、缓存、测量系统、文件服务、Web服务等内容，应用程序开发者只需要使用SparkContext提供的API完成功能开发。

SparkContext内置的DAGScheduler负责创建Job，将DAG中的RDD划分到不同的Stage，提交Stage等功能。
Spark的应用程序的编写方式： 是一种链式调用
DAGScheduler的角色是针对调用链执行划分，划分的标准是Shuffle

SparkContext内置的TaskScheduler负责资源的申请、任务的提交及请求集群对任务的调度等工作。
TaskScheduler就是帮stage然后变成了taskset

2、存储体系

Spark优先考虑使用各节点的内存作为存储，当内存不足时才会考虑使用磁盘，这极大地减少了磁盘I/O，提升了任务执行的效率，使得Spark适用于实时计算、流式计算等场景。

小总结：如果一个Spark应用程序中的各个stage的数据，全部下放到磁盘，就其实和MapReduce区别不大了。

此外，Spark还提供了以内存为中心的高容错的分布式文件系统Tachyon（现在改名为Alluxio）供用户进行选择。Tachyon能够为Spark提供可靠的内存级的文件共享服务。

3、计算引擎

计算引擎由SparkContext中的DAGScheduler、RDD以及具体节点上的Executor负责执行的Map和Reduce任务组成。

DAGScheduler和RDD虽然位于SparkContext内部，但是在任务正式提交与执行之前将Job中的RDD组织成有向无环图（简称DAG）、并对Stage进行划分决定了任务执行阶段任务的数量、迭代计算、shuffle等过程。

4、部署模式

由于单节点不足以提供足够的存储及计算能力，所以作为大数据处理的Spark在SparkContext的TaskScheduler组件中提供了对Standalone部署模式的实现和YARN、Mesos等分布式资源管理系统的支持。

通过使用Local、Standalone、YARN、Mesos、kubernetes、Cloud等部署模式为Task分配计算资源，提高任务的并发执行效率。除了可用于实际生产环境的Standalone、YARN、Mesos、kubernetes、Cloud等部署模式外，Spark还提供了Local模式和local-cluster模式便于开发和调试。

4.2 Spark应用模块

Spark的底层提供了一个基于内存/磁盘的批处理计算引擎。spark基于这个引擎，提供了很多高级的应用模块。解决了不同场景中的业务需求。

为了扩大应用范围，Spark陆续增加了一些扩展功能，主要包括：

1、Spark SQL

spark sql 相对于 spark core 就类似于 Hive 相对于 MapReduce。
由于SQL具有普及率高、学习成本低等特点，为了扩大Spark的应用面，因此增加了对SQL及Hive的支持。Spark SQL的过程可以总结为：首先使用SQL语句解析器（SqlParser）将SQL转换为语法树（Tree），并且使用规则执行器（RuleExecutor）将一系列规则（Rule）应用到语法树，最终生成物理执行计划并执行的过程。其中，规则包括语法分析器（Analyzer）和优化器（Optimizer）。Hive的执行过程与SQL类似。

spark sql相对于spark core 就类似于Hive相对于mapreduce ， 用来处理结构化数据。

2、Spark Streaming

Spark Streaming与Apache Storm类似，也用于流式计算。
	
Spark Streaming支持Kafka、Flume、Twitter、MQTT、ZeroMQ、Kinesis和简单的TCP套接字等多种数据输入源。输入流接收器（Receiver）负责接入数据，是接入数据流的接口规范。

Dstream是Spark Streaming中所有数据流的抽象，Dstream可以被组织为StreamGraph。Dstream本质上由一系列连续的RDD组成。

3、Spark GraphX

Spark提供的分布式图计算框架。GraphX主要遵循整体同步并行计算模式（BulkSynchronous Parallell，简称BSP）下的Pregel模型实现。

GraphX提供了对图的抽象Graph，Graph由顶点（Vertex）、边（Edge）及继承了Edge的EdgeTriplet（添加了srcAttr和dstAttr用来保存源顶点和目的顶点的属性）三种结构组成。

GraphX目前已经封装了最短路径、网页排名、连接组件、三角关系统计等算法的实现，用户可以选择使用。

4、Spark MLlib

Spark提供的机器学习框架。机器学习是一门涉及概率论、统计学、逼近论、凸分析、算法复杂度理论等多领域的交叉学科。MLlib目前已经提供了基础统计、分类、回归、决策树、随机森林、朴素贝叶斯、保序回归、协同过滤、聚类、维数缩减、特征提取与转型、频繁模式挖掘、预言模型标记语言、管道等多种数理统计、概率论、数据挖掘方面的数学算法。

4.3 Spark基本架构

Spark的架构分为两个方面：

集群的架构：  master    worker    主从架构    JVM进程
应用程序架构：driver    executor       JVM进程

应用程序由一个driver和n多个executor组成
在standalone集群模式中，其实master就是cluster Manager
executor：任务执行进程
任务进程会初始化一个线程池
线程池中的每个线程会运行一个 task

从应用程序运行的角度来看，Spark集群由以下部分组成

集群模式链接： http://spark.apache.org/docs/latest/cluster-overview.html

Cluster Manager

Spark的集群管理器，主要负责资源的分配与管理。集群管理器分配的资源属于一级分配，它将各个Worker上的内存、CPU等资源分配给应用程序，但是并不负责对Executor的资源分配。目前，Standalone、YARN、Mesos、K8S，EC2等都可以作为Spark的集群管理器。

Master

Spark集群的主节点。负责管理整个集群

Worker

Spark集群的工作节点。对Spark应用程序来说，由集群管理器分配得到资源的Worker节点主要负责以下工作：
创建Executor，将资源和任务进一步分配给Executor，同步资源信息给Cluster Manager。

Executor

执行计算任务的一些进程。主要负责任务的执行以及与Worker、Driver Application的信息同步。

Driver Appication

客户端驱动程序，也可以理解为客户端应用程序，用于将任务程序转换为RDD和DAG，并与Cluster Manager进行通信与调度。

这些组成部分之间的整体关系如下图所示：

Spark计算平台有两个重要角色，Driver和executor，不论是StandAlone模式还是YARN模式，都是Driver充当Application的master角色，负责任务执行计划生成和任务分发及调度；executor充当worker角色，负责实际执行任务的task，计算的结果返回Driver。

4.4 Spark核心概念

Term	Meaning
Application	User program built on Spark. Consists of a driver program and executors on the cluster.
Application jar	A jar containing the user’s Spark application. In some cases users will want to create an “uber jar” containing their application along with its dependencies. The user’s jar should never include Hadoop or Spark libraries, however, these will be added at runtime.
Driver program	The process running the main() function of the application and creating the SparkContext
Cluster manager	An external service for acquiring resources on the cluster (e.g. standalone manager, Mesos, YARN, Kubernetes)
Deploy mode	Distinguishes where the driver process runs. In “cluster” mode, the framework launches the driver inside of the cluster. In “client” mode, the submitter launches the driver outside of the cluster.
Worker node	Any node that can run application code in the cluster
Executor	A process launched for an application on a worker node, that runs tasks and keeps data in memory or disk storage across them. Each application has its own executors.
Task	A unit of work that will be sent to one executor
Job	A parallel computation consisting of multiple tasks that gets spawned in response to a Spark action (e.g. save, collect); you’ll see this term used in the driver’s logs.
Stage	Each job gets divided into smaller sets of tasks called stages that depend on each other (similar to the map and reduce stages in MapReduce); you’ll see this term used in the driver’s logs.

Spark中的核心概念文字描述：

1、Application：表示你的应用程序，包含一个Driver Program和若干Executor

2、Driver Program：Spark中的Driver即运行上述Application的main()函数并且创建SparkContext，其中创建SparkContext的目的是为了准备Spark应用程序的运行环境。由SparkContext负责与ClusterManager通信，进行资源的申请，任务的分配和监控等。程序执行完毕后关闭SparkContext

3、ClusterManager：在Standalone模式中即为Master（主节点），控制整个集群，监控Worker。在YARN模式中为资源管理器。

4、SparkContext：整个应用的上下文，控制应用程序的生命周期，负责调度各个运算资源，协调各个Worker上的Executor。初始化的时候，会初始化DAGScheduler和TaskScheduler两个核心组件。

5、RDD：Spark的基本计算单元，一组RDD可形成执行的有向无环图RDD Graph。

6、DAGScheduler：根据Job构建基于Stage的DAG，并提交Stage给TaskScheduler，其划分Stage的依据是RDD之间的依赖关系：宽依赖，也叫shuffle依赖

7、TaskScheduler：将TaskSet提交给Worker（集群）运行，每个Executor运行什么Task就是在此处分配的。

8、Worker：集群中可以运行Application代码的节点。在Standalone模式中指的是通过worker文件配置的worker节点，在Spark on Yarn模式中指的就是NodeManager节点。

9、Executor：某个Application运行在Worker节点上的一个进程，该进程负责运行某些task，并且负责将数据存在内存或者磁盘上。在Spark on Yarn模式下，其进程名称为 CoarseGrainedExecutorBackend，一个CoarseGrainedExecutorBackend进程有且仅有一个executor对象，它负责将Task包装成taskRunner，并从线程池中抽取出一个空闲线程运行Task，这样，每个CoarseGrainedExecutorBackend能并行运行Task的数据就取决于分配给它的CPU的个数。

10、Stage：每个Job会被拆分很多组Task，每组作为一个TaskSet，其名称为Stage

11、Job：包含多个Task组成的并行计算，是由Action行为触发的

12、Task：在Executor进程中执行任务的工作单元，多个Task组成一个Stage

13、SparkEnv：线程级别的上下文，存储运行时的重要组件的引用。SparkEnv内创建并包含如下一些重要组件的引用。
	MapOutPutTracker：负责Shuffle元信息的存储。
	BroadcastManager：负责广播变量的控制与元信息的存储。
	BlockManager：负责存储管理、创建和查找块。
	MetricsSystem：监控运行时性能指标信息。
	SparkConf：负责存储配置信息

4.5 编程模型

Spark 应用程序的编程流程：
1、初始化创建一个sparkContext编程入口
2、通过编程入口sparkContext加载数据源获取数据抽象对象(RDD)
3、针对这个RDD链式调用各种算子执行各种逻辑操作得到一个结果RDD
4、针对执行结果RDD执行落盘操作
5、提交Job任务运行。

Spark应用程序从编写到提交、执行、输出的整个过程如下图所示：

图中描述的步骤如下：

1、用户使用SparkContext提供的API（常用的有textFile、sequenceFile、runJob、stop等）编写Driver Application程序。此外SQLContext、HiveContext及StreamingContext对SparkContext进行封装，并提供了SQL、Hive及流式计算相关的API。

2、使用SparkContext提交的用户应用程序，首先会使用BlockManager和BroadcastManager将任务的Hadoop配置进行广播。然后由DAGScheduler将任务转换为RDD并组织成DAG，DAG还将被划分为不同的Stage。最后由TaskScheduler借助ActorSystem将任务提交给集群管理器（ClusterManager）。

3、集群管理器（ClusterManager）给任务分配资源，即将具体任务分配到Worker上，Worker创建Executor来处理任务的运行。Standalone、YARN、Mesos、kubernetes、EC2等都可以作为Spark的集群管理器。

计算模型：

​ RDD可以看做是对各种数据计算模型的统一抽象，Spark的计算过程主要是RDD的迭代计算过程，如上图。RDD的迭代计算过程非常类似于管道。分区数量取决于partition数量的设定，每个分区的数据只会在一个Task中计算。所有分区可以在多个机器节点的Executor上并行执行。

声明：
        文章中代码及相关语句为自己根据相应理解编写，文章中出现的相关图片为自己实践中的截图和相关技术对应的图片，若有相关异议，请联系删除。感谢。转载请注明出处，感谢。

By luoyepiaoxue2014

B站： https://space.bilibili.com/1523287361 点击打开链接
微博地址： http://weibo.com/luoyepiaoxue2014 点击打开链接

转载Spark系列之运行原理和架构

参考 http://www.cnblogs.com/shishanyuan/p/4721326.html

 

 1、 Spark运行架构

1.1 术语定义

lApplication：Spark Application的概念和Hadoop MapReduce中的类似，指的是用户编写的Spark应用程序，
包含了一个Driver 功能的代码和分布在集群中多个节点上运行的Executor代码；

lDriver：Spark中的Driver即运行上述Application的main()函数并且创建SparkContext，
其中创建SparkContext的目的是为了准备Spark应用程序的运行环境。
在Spark中由SparkContext负责和ClusterManager通信，进行资源的申请、任务的分配和监控等；
当Executor部分运行完毕后，Driver负责将SparkContext关闭。通常用SparkContext代表Drive；

lExecutor：Application运行在Worker 节点上的一个进程，该进程负责运行Task，并且负责将数据存在内存或者磁盘上，
每个Application都有各自独立的一批Executor。
在Spark on Yarn模式下，其进程名称为CoarseGrainedExecutorBackend，类似于Hadoop MapReduce中的YarnChild。
一个CoarseGrainedExecutorBackend进程有且仅有一个executor对象，它负责将Task包装成taskRunner，并从线程池中抽取出一个空闲线程运行Task。
每个CoarseGrainedExecutorBackend能并行运行Task的数量就取决于分配给它的CPU的个数了；

lCluster Manager：指的是在集群上获取资源的外部服务，目前有：
Ø  Standalone：Spark原生的资源管理，由Master负责资源的分配；
Ø  Hadoop Yarn：由YARN中的ResourceManager负责资源的分配；

lWorker：集群中任何可以运行Application代码的节点，类似于YARN中的NodeManager节点。
在Standalone模式中指的就是通过Slave文件配置的Worker节点，在Spark on Yarn模式中指的就是NodeManager节点；

l作业（Job）：包含多个Task组成的并行计算，往往由Spark Action催生，一个JOB包含多个RDD及作用于相应RDD上的各种Operation；

l阶段（Stage）：每个Job会被拆分很多组Task，每组任务被称为Stage，也可称TaskSet，一个作业分为多个阶段；

l任务（Task）： 被送到某个Executor上的工作任务；

1.2 Spark运行基本流程

Spark运行基本流程参见下面示意图

1.   构建Spark Application的运行环境（启动SparkContext），SparkContext向资源管理器（可以是Standalone、Mesos或YARN）注册并申请运行Executor资源；

2.   资源管理器分配Executor资源并启动StandaloneExecutorBackend，Executor运行情况将随着心跳发送到资源管理器上；

3.   SparkContext构建成DAG图，将DAG图分解成Stage，并把Taskset发送给Task Scheduler。Executor向SparkContext申请Task，Task Scheduler将Task发放给Executor运行同时SparkContext将应用程序代码发放给Executor。

4.   Task在Executor上运行，运行完毕释放所有资源。

Spark运行架构特点：

l每个Application获取专属的executor进程，该进程在Application期间一直驻留，并以多线程方式运行tasks。
这种Application隔离机制有其优势的，无论是从调度角度看（每个Driver调度它自己的任务），
还是从运行角度看（来自不同Application的Task运行在不同的JVM中）。
当然，这也意味着Spark Application不能跨应用程序共享数据，除非将数据写入到外部存储系统。

lSpark与资源管理器无关，只要能够获取executor进程，并能保持相互通信就可以了。

l提交SparkContext的Client应该靠近Worker节点（运行Executor的节点)，最好是在同一个Rack里，
因为Spark Application运行过程中SparkContext和Executor之间有大量的信息交换；
如果想在远程集群中运行，最好使用RPC将SparkContext提交给集群，不要远离Worker运行SparkContext。

lTask采用了数据本地性和推测执行的优化机制。

 

 

1.2.1 DAGScheduler

DAGScheduler把一个Spark作业转换成Stage的DAG（Directed Acyclic Graph有向无环图），根据RDD和Stage之间的关系找出开销最小的调度方法，然后把Stage以TaskSet的形式提交给TaskScheduler。

 

1.2.2 TaskScheduler

DAGScheduler决定了运行Task的理想位置，并把这些信息传递给下层的TaskScheduler。此外，DAGScheduler还处理由于Shuffle数据丢失导致的失败，这有可能需要重新提交运行之前的Stage（非Shuffle数据丢失导致的Task失败由TaskScheduler处理）。

TaskScheduler维护所有TaskSet，当Executor向Driver发送心跳时，TaskScheduler会根据其资源剩余情况分配相应的Task。另外TaskScheduler还维护着所有Task的运行状态，重试失败的Task。下图展示了TaskScheduler的作用：

在不同运行模式中任务调度器具体为：

l  Spark on Standalone模式为TaskScheduler；

l  YARN-Client模式为YarnClientClusterScheduler

l  YARN-Cluster模式为YarnClusterScheduler

 

1.3 RDD运行原理

那么 RDD在Spark架构中是如何运行的呢？总高层次来看，主要分为三步：

1.创建 RDD 对象

2.DAGScheduler模块介入运算，计算RDD之间的依赖关系。RDD之间的依赖关系就形成了DAG

3.每一个JOB被分为多个Stage，划分Stage的一个主要依据是当前计算因子的输入是否是确定的，如果是则将其分在同一个Stage，避免多个Stage之间的消息传递开销。

 

以下面一个按 A-Z 首字母分类，查找相同首字母下不同姓名总个数的例子来看一下 RDD 是如何运行起来的。

 

步骤 1 ：创建 RDD  上面的例子除去最后一个 collect 是个动作，不会创建 RDD 之外，前面四个转换都会创建出新的RDD 。因此第一步就是创建好所有 RDD( 内部的五项信息 ) 。

步骤 2 ：创建执行计划 Spark 会尽可能地管道化，并基于是否要重新组织数据来划分 阶段 (stage) ，例如本例中的groupBy() 转换就会将整个执行计划划分成两阶段执行。最终会产生一个 DAG(directed acyclic graph ，有向无环图 ) 作为逻辑执行计划。

步骤 3 ：调度任务  将各阶段划分成不同的 任务 (task) ，每个任务都是数据和计算的合体。在进行下一阶段前，当前阶段的所有任务都要执行完成。因为下一阶段的第一个转换一定是重新组织数据的，所以必须等当前阶段所有结果数据都计算出来了才能继续。

假设本例中的 hdfs://names 下有四个文件块，那么 HadoopRDD 中 partitions 就会有四个分区对应这四个块数据，同时 preferedLocations 会指明这四个块的最佳位置。现在，就可以创建出四个任务，并调度到合适的集群结点上。

 

 

 

2、Spark在不同集群中的运行架构

Spark注重建立良好的生态系统，它不仅支持多种外部文件存储系统，提供了多种多样的集群运行模式。部署在单台机器上时，既可以用本地（Local）模式运行，也可以使用伪分布式模式来运行；当以分布式集群部署的时候，可以根据自己集群的实际情况选择Standalone模式（Spark自带的模式）、YARN-Client模式或者YARN-Cluster模式。

Spark的各种运行模式虽然在启动方式、运行位置、调度策略上各有不同，但它们的目的基本都是一致的，就是在合适的位置安全可靠的根据用户的配置和Job的需要运行和管理Task。

 

 

2.1 Spark on Standalone运行过程

Standalone模式是Spark实现的资源调度框架，其主要的节点有Client节点、Master节点和Worker节点。其中Driver既可以运行在Master节点上中，也可以运行在本地Client端。

当用spark-shell交互式工具提交Spark的Job时，Driver在Master节点上运行；

当使用spark-submit工具提交Job或者在Eclips、IDEA等开发平台上使用”new SparkConf.setManager(“spark://master:7077”)”方式运行Spark任务时，Driver是运行在本地Client端上的。

 

 

2.2  Spark on YARN运行过程

YARN是一种统一资源管理机制，在其上面可以运行多套计算框架。目前的大数据技术世界，大多数公司除了使用Spark来进行数据计算，由于历史原因或者单方面业务处理的性能考虑而使用着其他的计算框架，比如MapReduce、Storm等计算框架。Spark基于此种情况开发了Spark on YARN的运行模式，由于借助了YARN良好的弹性资源管理机制，不仅部署Application更加方便，而且用户在YARN集群中运行的服务和Application的资源也完全隔离，更具实践应用价值的是YARN可以通过队列的方式，管理同时运行在集群中的多个服务。

Spark on YARN模式根据Driver在集群中的位置分为两种模式：一种是YARN-Client模式，另一种是YARN-Cluster（或称为YARN-Standalone模式）。

2.2.1 YARN框架流程

任何框架与YARN的结合，都必须遵循YARN的开发模式。在分析Spark on YARN的实现细节之前，有必要先分析一下YARN框架的一些基本原理。

Yarn框架的基本运行流程图为：

 

其中，ResourceManager负责将集群的资源分配给各个应用使用，而资源分配和调度的基本单位是Container，其中封装了机器资源，如内存、CPU、磁盘和网络等，每个任务会被分配一个Container，该任务只能在该Container中执行，并使用该Container封装的资源。

NodeManager是一个个的计算节点，主要负责启动Application所需的Container，监控资源（内存、CPU、磁盘和网络等）的使用情况并将之汇报给ResourceManager。

ResourceManager与NodeManagers共同组成整个数据计算框架，ApplicationMaster与具体的Application相关，主要负责同ResourceManager协商以获取合适的Container，并跟踪这些Container的状态和监控其进度。

 

2.2.2 YARN-Client

 

2.2.3 YARN-Cluster

在YARN-Cluster模式中，当用户向YARN中提交一个应用程序后，YARN将分两个阶段运行该应用程序：第一个阶段是把Spark的Driver作为一个ApplicationMaster在YARN集群中先启动；第二个阶段是由ApplicationMaster创建应用程序，然后为它向ResourceManager申请资源，并启动Executor来运行Task，同时监控它的整个运行过程，直到运行完成。

YARN-cluster的工作流程分为以下几个步骤：

 

2.2.4 YARN-Client 与 YARN-Cluster 区别

理解YARN-Client和YARN-Cluster深层次的区别之前先清楚一个概念：Application Master。在YARN中，每个Application实例都有一个ApplicationMaster进程，它是Application启动的第一个容器。它负责和ResourceManager打交道并请求资源，获取资源之后告诉NodeManager为其启动Container。从深层次的含义讲YARN-Cluster和YARN-Client模式的区别其实就是ApplicationMaster进程的区别。

l  YARN-Client模式下，Driver运行在Client上，Application Master仅仅向YARN请求Executor，Client会和请求的Container通信来调度他们工作，也就是说Client不能离开。

l  YARN-Cluster模式下，Driver运行在AM(Application Master)中，它负责向YARN申请资源，并监督作业的运行状况。当用户提交了作业之后，就可以关掉Client，作业会继续在YARN上运行，因而YARN-Cluster模式不适合运行交互类型的作业；

 

下面还有一些不同模式下的运行例子。略。

其中有一个 xxx.cache() ，意思是把运算的结果放在内存里。

 

参考资料：

（1）《Spark1.0.0 运行架构基本概念》 http://blog.csdn.net/book_mmicky/article/details/25714419

（2）《Spark架构与作业执行流程简介》 http://www.cnblogs.com/shenh062326/p/3658543.html

（3）《Spark1.0.0 运行架构基本概念》 http://shiyanjun.cn/archives/744.html

 

（完）