《Spark编程基础及项目实践》课件03Spark的设计与运行原理

项目3Spark的设计与运行原理学习目标（1）（2）（3）了解Spark架构中各个组件的作用熟悉Spark的运行流程掌握Spark的工作机制学习目标任务1Spark运行架构3.1.1Spark架构设计Spark运行架构如图3-1所示，包括应用程序的控制进程DriverProgram、资源管理器ClusterManager、工作节点Worker和执行进程Executor。DriverProgram进行任务控制，ClusterManager负责资源分配，Worker用于运行多个Executor，Executor负责具体的计算和数据存储，每个Executor又由多个具体的Task（任务）组成。图3-1Spark运行架构3.1.1Spark架构设计关于Spark架构的组件介绍如下。（1）Application用户编写的Spark应用程序。由一个DriverProgram代码和分布在集群上的一个或多个Executor代码构成。（2）DriverProgram运行Application的main函数并创建SparkContext的控制进程，负责任务控制。（3）SparkContext应用上下文对象。SparkContext会准备Spark应用程序的运行环境，向ClusterManager申请资源，进行任务的分配和监控等。3.1.1Spark架构设计关于Spark架构的组件介绍如下。（4）ClusterManger资源管理器。在集群上获取资源的外部服务，用于进行资源调度和监控资源的使用情况，主要有3种：Standalone、Mesos和YARN。（5）Worker工作节点。在集群中运行应用程序代码的节点，一个Worker可以运行一个或多个Executor进程。（6）Executor执行进程。运行在Worker节点上的应用程序进程，负责运行Task和存储数据，它可以将数据存储在内存或磁盘上。（7）Task：任务分配给Executor运行的工作单元，一个Executor可以运行一个或多个Task。3.1.2

Spark运行流程在学习Spark运行流程之前，先来认识几个重要的概念。一个由多个Task组成的并行计算，执行SparkAction算子时会触发Job。（1）Job：作业看这里！每个Job被划分成了多个相互独立的Stage，每个Stage由一组Task组成。类似于MapReduce中的Map和Reduce。在Spark中，一个Application由一个含有入口函数main的DriverProgram和一个或多个Job构成，每个Job包含了多个Stage，每个Stage由多个Task组成。（2）Stage：阶段RDD：弹性分布式数据集（resilientdistributeddataset）的英文缩写，一个RDD就是一个分布式对象集合，它是Spark的基本计算单元。在进行数据存储时，首先将数据放入内存，当内存放不下时会将一部分数据放到磁盘。（3）RDD3.1.2

Spark运行流程在学习Spark运行流程之前，先来认识几个重要的概念。有向无环图，Spark中使用DAG对RDD的关系进行建模。（4）DAG看这里！SparkContext将Application拆分成很多的Job，每个Job根据RDD之间的依赖关系构建成DAG，然后提交给DAGScheduler。DAGScheduler将DAG划分为多个Stage，每个Stage都是一个TaskSet（任务集）。（5）DAGScheduler3.1.2

Spark运行流程在学习Spark运行流程之前，先来认识几个重要的概念。将TaskSet分发给Worker运行。（6）TaskScheduler看这里！作用在DriverProgram中，负责分配任务。（7）SchedulerBackend作用在Executor中，管理所有任务的运行，接收到SchedulerBackend分配的任务后，将任务派发给线程池中的空闲线程去执行。ExecutorBackend和Executor是一一对应的。（8）ExecutorBackend3.1.2

Spark运行流程Spark的基本运行流程如图3-2所示，基本运行流程如下。图3-2Spark的基本运行流程3.1.2

Spark运行流程（1）初始化Spark应用程序提交后，首先启动DriverProgram运行Application的main函数并创建SparkContext。SparkContext初始化时会初始化DAGScheduler和TaskScheduler两个核心组件。（2）SparkContext连接到ClusterManager，向ClusterManger注册并申请资源，如申请内存和磁盘等资源。ClusterManager目前主要有Spark原生的资源管理器Standalone、Mesos或YARN。（3）Worker节点会定期向ClusterManager发送心跳信息，这些心跳信息中包含Executor的状态。ClusterManager根据Worker节点发送的心跳信息和SparkContext申请的资源情况来决定使用哪些Worker节点，并将资源分配给这些Worker，启动Executor，Executor初始化线程池。3.1.2

Spark运行流程（4）ExecutorBackend向SparkContext注册，SparkContext获取了它自己的Executor进程集。（5）SparkContext接收到用户代码后生成具体任务集的过程如下。②划分Stage①构建DAG③分发TaskSet3.1.2

Spark运行流程（6）ExecutorBackend接收到SchedulerBackend分配的任务后，将任务派发给线程池中的空闲线程去执行，Executor以多线程的方式并行执行任务，并将运行结果报告给SparkContext。（7）Task执行结束后，通过BlockManager将结果存储到内存或磁盘中。（8）注销。所有Task完成后，SparkContext向ClusterManager申请注销，释放所有资源。任务2Spark工作机制3.2.1

Spark内存管理1.内存划分Spark的一个重要特点是：它是基于内存计算的，因而更快。Executor是一个JVM进程，它的内存管理建立在JVM的内存管理之上。Spark采用堆内内存（onheapmemory）和堆外内存（offheapmemory）的规划机制，为了充分利用内存，Spark对堆内内存进行了详细的规划式管理，同时在Worker的系统内存中开辟空间作为堆外内存来进一步优化内存的使用。3.2.1

Spark内存管理1.内存划分堆内内存和堆外内存示意图如图3-3所示。图3-3Spark堆内内存和堆外内存示意图3.2.1

Spark内存管理1.内存划分堆内内存包括以下4块，默认情况下，Spark仅使用堆内内存。（1）ExecutionMemory01020304（2）StorageMemory（3）UserMemory（4）ReservedMemory3.2.1

Spark内存管理2.内存空间分配Spark1.6之前采用的是静态内存管理机制，即Execution内存、Storage内存和其他内存的分配是静态的，用户在Spark应用启动之前进行配置，在Spark应用运行期间，这些内存都是固定的。这就导致对内存的调优难以进行，Execution内存或Storage内存任何一方内存不足时，即使另一方有很大空闲空间也无法申请使用，内存资源利用率不高。3.2.1

Spark内存管理2.内存空间分配堆内内存的统一管理示意图如图3-4所示。图3-4堆内内存的统一管理示意图3.2.1

Spark内存管理2.内存空间分配SystemMemory（系统内存）就是通过参数executormemory或spark.executor.memory来设置堆内内存的。ReservedMemory（预留内存）默认为300MB，用于防止实际内存超出预设值而导UnifiedMemory（统一内存）为Execution内存和Storage内存共用的内存空间，约占UsableMemory的60%，具体占用的比例由参数spark.memory.fraction进行配置。3.2.1

Spark内存管理2.内存空间分配堆外内存的统一管理示意图如图3-5所示。图3-5堆外内存的统一管理示意图Storage内存和Execution空间大小由参数spark.memory.storageFraction来设定各自占用的比例，默认为各占off-heap内存的50%。堆外内存的动态占用机制与堆内内存一致。3.2.2

Spark容错机制

RDD之间的依赖关系有两种：窄依赖（narrowdependency）和宽依赖（widedependency），如图3-6所示。图3-6RDD的依赖关系1.Lineage机制3.2.2

Spark容错机制1.Lineage机制（1）窄依赖：一个父RDD的每个分区最多被一个子RDD的一个分区所使用，具体分为两种：一个子RDD分区只依赖于一个父RDD分区，一个子RDD分区依赖于多个父RDD分区。（2）宽依赖：一个父RDD至少有一个分区被一个子RDD的多个分区所使用。3.2.2

Spark容错机制2.Cache机制对于迭代型算法和交互式应用，可能需要对同一块数据进行多次重复访问，若每次都进行重新运算，则时间开销太大。Spark采用了Cache机制来保证需要访问重复数据的应用可以运行得更快。在计算RDD分区时判断这个分区要不要进行缓存，如果设置了需要进行缓存，那么先将分区计算出来，然后通过BlockManager缓存到内存中。在下一次使用该RDD分区时，Task会先通过BlockManager去内存中查看是否有这个RDD的缓存，若有则直接读取使用而不再重新进行计算。Cache机制适用于那些会被重复使用的但是又不太大的RDD，而对于运算时间很长或运算量很大的RDD和计算链过长或依赖其他RDD很多的RDD，则需要使用Checkpoint机制。3.2.2

Spark容错机制3.Checkpoint机制RDD窄依赖的重新计算代价比较小，但是对于RDD宽依赖来说，重新计算的代价是非常大的。如图3-7所示，左图是窄依赖，右图是宽依赖，如果b1分区丢失，对于左图来说，只需要重新计算a1便可得到b1，而对于右图来说，需要将a1、a2、a3和a4都重新计算，这就产生了冗余计算，因为计算a1、a2、a3和a4相当于把b2、b3和b4也重新计算了。图3-7窄依赖和宽依赖的重新计算3.2.3

Spark监控管理Spark提供了3种方式来对应用进行监控：010203WebUI指标第三方监控工具3.2.3

Spark监控管理例如，使用sparkshell提交应用，用于统计文件“source.md”中包含“Spark”关键字的行数，代码如下。valtextFile=spark.read.textFile("file:///hadoop/source.md")textFile.counttextFile.first()vallinesWithSpark=textFile.filter(line=>line.contains("Spark"))linesWithSpark.cache()//设置缓存linesWithSpark.count()linesWithSpark.count()程序运行后，在浏览器中访问“http://master:4040”就可以查看程序的监控信息。3.2.3

Spark监控管理如图3-8所示，可以看到有4个Job。图3-8（a）展示了事件发生的时间线，从中可以直观地看到每个步骤的时间消耗情况，如果某个步骤的时间消耗异常，就可以有针对性地对这个步骤进行调试。每个Job的具体提交时间、运行时长、Stage和Task的个数及完成情况都可以在图3-8（b）中看到。图中用方框标示的部分为通过WebUI可以看到的选项。图3-8Job监控3.2.3

Spark监控管理图3-9所示为Stage的监控信息，共有7个Stage，每个Stage的提交时间、运行时长、Task的个数与完成情况、Task输入/输出文件大小及ShuffleWrite大小都清晰地显示出来。图3-9Stage监控3.2.3

Spark监控管理单击Description可以看到Stage的详细信息，图3-10展示了Stage0的DAG，图3-11展示了Stage0的Task和Executor的详细信息。图3-10Stage0的DAG图3-11Stage0的Task和Executor的详细信息3.2.3

Spark监控管理图3-12展示的是Storage监控中某个RDD的缓存信息，该RDD占用561.9KB的堆内内存，具有1个分区。图3-12某个RDD的缓存信息3.2.3

Spark监控管理图3-13展示了Executor的运行情况。图3-13Executor的运行情况小结Spark的运行架构包括应用程序的控制进程DriverProgram、资源管理器ClusterManager、工作节点Worker和执行进程Executor。SparkContext接收到用户代码后会执行以下操作：构建DAG、划分Stage、分发TaskSet、Executor以多线程方式并行执行任务、注销。Spark是基于内存计算的，采用堆内内存和堆外内存的规划机制。Spark1.6之后引入了统一内存管理机制，Execution内存和Storage内存共享同一块空间，互相可以动态调整。Spark采用的RDD具有高效的容错性，主要使用了Lineage、Cache和Checkpoint三种方式。Spark提供了3种方式来对应用进行监控：WebUI、指标和第三方监控工具，其中WebUI监控是默认开启的，也是最常用的方式。小结习题1.填空题（1）Spark的运行架构包括应用程序的

、

、和

。（2）Spark的基本计算单元是

。（3）Spark中使用

对RDD的关系进行建模。（4）Spark的一个重要特点是基于

计算的，因而更快。（5）Spark采用

和

的规划机制。（6）Spark1.6之后引入了统一内存管理机制，

和

共享同一块空间，互相可以动态调整。习题2.选择题（1）在Spark运行架构中，以下（