工作原理_一文理解Spark基础概念及工作原理

文章作者&＃xff1a;熊峰&＃xff0c;腾讯大数据研发工程师。

一、Spark 介绍及生态

Spark是UC Berkeley AMP Lab开源的通用分布式并行计算框架&＃xff0c;目前已成为Apache软件基金会的顶级开源项目。至于为什么我们要学习Spark&＃xff0c;可以总结为下面三点&＃xff1a;

1. Spark相对于hadoop的优势

&＃xff08;1&＃xff09;高性能

Spark具有hadoop MR所有的优点&＃xff0c;hadoop MR每次计算的中间结果都会存储到HDFS的磁盘上&＃xff0c;而Spark的中间结果可以保存在内存&＃xff0c;在内存中进行数据处理。

&＃xff08;2&＃xff09;高容错

• 基于“血统”(Lineage)的数据恢复&＃xff1a;spark引入了弹性分布式数据集RDD的抽象&＃xff0c;它是分布在一组节点中的只读的数据的集合&＃xff0c;这些集合是弹性的且是相互依赖的&＃xff0c;如果数据集中的一部分的数据发生丢失可以根据“血统”关系进行重建。

• CheckPoint容错&＃xff1a;RDD计算时可以通过checkpoint进行容错&＃xff0c;checkpoint有两种检测方式&＃xff1a;通过冗余数据和日志记录更新操作。在RDD中的doCheckPoint方法相当于通过冗余数据来缓存数据&＃xff0c;而“血统”是通过粗粒度的记录更新操作来实现容错的。CheckPoint容错是对血统检测进行的容错辅助&＃xff0c;避免“血统”(Lineage)过长造成的容错成本过高。

&＃xff08;3&＃xff09;spark的通用性

spark 是一个通用的大数据计算框架&＃xff0c;相对于hadoop它提供了更丰富的使用场景。

spark相对于hadoop map reduce两种操作还提供了更为丰富的操作&＃xff0c;分为action(collect,reduce,save…)和transformations&＃xff08;map,union,join,filter…&＃xff09;,同时在各节点的通信模型中相对于hadoop的shuffle操作还有分区&＃xff0c;控制中间结果存储&＃xff0c;物化视图等。

2. spark 生态介绍

Spark支持多种编程语言&＃xff0c;包括Java、Python、R和Scala。在计算资源调度层支持local模式&＃xff0c;standalone模式&＃xff0c;yarn模式以及k8s等。

同时spark有多组件的支持应用场景&＃xff0c;在spark core的基础上提供了spark Streaming&＃xff0c;spark SQL&＃xff0c;spark Mllib&＃xff0c;spark R&＃xff0c;GraphX等组件。

spark Streaming用于实时流计算&＃xff0c;spark SQL旨在将熟悉的SQL数据库查询与更复杂的基于算法的分析相结合&＃xff0c;GraphX用于图计算&＃xff0c;spark Mllib用于机器学习&＃xff0c;spark R用于对R语言的数据计算。

spark 支持多种的存储介质&＃xff0c;在存储层spark支持从hdfs&＃xff0c;hive&＃xff0c;aws等读入和写出数据&＃xff0c;也支持从hbase&＃xff0c;es等大数据库中读入和写出数据&＃xff0c;同时也支持从mysql&＃xff0c;pg等关系型数据库中读入写出数据&＃xff0c;在实时流计算在可以从flume&＃xff0c;kafka等多种数据源获取数据并执行流式计算。

在数据格式上spark也支持的非常丰富&＃xff0c;比如常见的txt&＃xff0c;json&＃xff0c;csv等格式。同时也支持parquet&＃xff0c;orc&＃xff0c;avro等格式&＃xff0c;这几种格式在数据压缩和海量数据查询上优势也较为明显。

二、spark 原理及特点

1. spark core

Spark Core是Spark的核心&＃xff0c;其包含如下几个部分&＃xff1a;

&＃xff08;1&＃xff09;spark 基础配置

sparkContext是spark应用程序的入口&＃xff0c;spark应用程序的提交和执行离不开sparkContext&＃xff0c;它隐藏了网络通信&＃xff0c;分布式部署&＃xff0c;消息通信&＃xff0c;存储体系&＃xff0c;计算存储等&＃xff0c;开发人员只需要通过sparkContext等api进行开发即可。

sparkRpc 基于netty实现&＃xff0c;分为异步和同步两种方式。事件总线主要用于sparkContext组件间的交换&＃xff0c;它属于监听者模式&＃xff0c;采用异步调用。度量系统主要用于系统的运行监控。

&＃xff08;2&＃xff09;spark 存储系统

它用于管理spark运行中依赖的数据存储方式和存储位置&＃xff0c;spark的存储系统优先考虑在各节点以内存的方式存储数据&＃xff0c;内存不足时将数据写入磁盘中&＃xff0c;这也是spark计算性能高的重要原因。

我们可以灵活的控制数据存储在内存还是磁盘中&＃xff0c;同时可以通过远程网络调用将结果输出到远程存储中&＃xff0c;比如hdfs&＃xff0c;hbase等。

&＃xff08;3&＃xff09;spark 调度系统

spark 调度系统主要由DAGScheduler和TaskScheduler组成。

DAGScheduler 主要是把一个Job根据RDD间的依赖关系&＃xff0c;划分为多个Stage&＃xff0c;对于划分后的每个Stage都抽象为一个或多个Task组成的任务集&＃xff0c;并交给TaskScheduler来进行进一步的任务调度。而TaskScheduler 负责对每个具体的Task进行调度。

具体调度算法有FIFO&＃xff0c;FAIR&＃xff1a;

• FIFO调度&＃xff1a;先进先出&＃xff0c;这是Spark默认的调度模式。

• FAIR调度&＃xff1a;支持将作业分组到池中&＃xff0c;并为每个池设置不同的调度权重&＃xff0c;任务可以按照权重来决定执行顺序。

2. spark sql

spark sql提供了基于sql的数据处理方法&＃xff0c;使得分布式的数据集处理变的更加简单&＃xff0c;这也是spark 广泛使用的重要原因。

目前大数据相关计算引擎一个重要的评价指标就是&＃xff1a;是否支持sql&＃xff0c;这样才会降低使用者的门槛。spark sql提供了两种抽象的数据集合DataFrame和DataSet。

DataFrame 是spark Sql 对结构化数据的抽象&＃xff0c;可以简单的理解为spark中的表&＃xff0c;相比较于RDD多了数据的表结构信息(schema).DataFrame &＃61; Data &＃43; schema

RDD是分布式对象集合&＃xff0c;DataFrame是分布式Row的集合&＃xff0c;提供了比RDD更丰富的算子&＃xff0c;同时提升了数据的执行效率。

DataSet 是数据的分布式集合 &＃xff0c;它具有RDD强类型的优点 和Spark SQL优化后执行的优点。DataSet可以由jvm对象构建&＃xff0c;然后使用map&＃xff0c;filter&＃xff0c;flatmap等操作函数操作。

3. spark streaming

这个模块主要是对流数据的处理&＃xff0c;支持流数据的可伸缩和容错处理&＃xff0c;可以与Flume和Kafka等已建立的数据源集成。Spark Streaming的实现&＃xff0c;也使用RDD抽象的概念&＃xff0c;使得在为流数据编写应用程序时更为方便。

4. spark特点

&＃xff08;1&＃xff09;spark 计算速度快

spark将每个任务构建成DAG进行计算&＃xff0c;内部的计算过程通过弹性式分布式数据集RDD在内存在进行计算&＃xff0c;相比于hadoop的mapreduce效率提升了100倍。

&＃xff08;2&＃xff09;易于使用

spark 提供了大量的算子&＃xff0c;开发只需调用相关api进行实现无法关注底层的实现原理。

通用的大数据解决方案

相较于以前离线任务采用mapreduce实现&＃xff0c;实时任务采用storm实现&＃xff0c;目前这些都可以通过spark来实现&＃xff0c;降低来开发的成本。同时spark 通过spark SQL降低了用户的学习使用门槛&＃xff0c;还提供了机器学习&＃xff0c;图计算引擎等。

&＃xff08;3&＃xff09;支持多种的资源管理模式

学习使用中可以采用local 模型进行任务的调试&＃xff0c;在正式环境中又提供了standalone&＃xff0c;yarn等模式&＃xff0c;方便用户选择合适的资源管理模式进行适配。

&＃xff08;4&＃xff09;社区支持

spark 生态圈丰富&＃xff0c;迭代更新快&＃xff0c;成为大数据领域必备的计算引擎。

三、spark 运行模式及集群角色

1. spark运行模式

2. spark集群角色

下图是spark的集群角色图&＃xff0c;主要有集群管理节点cluster manager&＃xff0c;工作节点worker&＃xff0c;执行器executor&＃xff0c;驱动器driver和应用程序application 五部分组成&＃xff0c;下面详细说明每部分的特点。

&＃xff08;1&＃xff09;Cluster Manager

集群管理器&＃xff0c;它存在于Master进程中&＃xff0c;主要用来对应用程序申请的资源进行管理&＃xff0c;根据其部署模式的不同&＃xff0c;可以分为local&＃xff0c;standalone&＃xff0c;yarn&＃xff0c;mesos等模式。

&＃xff08;2&＃xff09;worker

worker是spark的工作节点&＃xff0c;用于执行任务的提交&＃xff0c;主要工作职责有下面四点&＃xff1a;

• worker节点通过注册机向cluster manager汇报自身的cpu&＃xff0c;内存等信息。

• worker 节点在spark master作用下创建并启用executor&＃xff0c;executor是真正的计算单元。

• spark master将任务Task分配给worker节点上的executor并执行运用。

• worker节点同步资源信息和executor状态信息给cluster manager。

在yarn 模式下运行worker节点一般指的是NodeManager节点&＃xff0c;standalone模式下运行一般指的是slave节点。

&＃xff08;3&＃xff09;executor

executor 是真正执行计算任务的组件&＃xff0c;它是application运行在worker上的一个进程。这个进程负责Task的运行&＃xff0c;它能够将数据保存在内存或磁盘存储中&＃xff0c;也能够将结果数据返回给Driver。

&＃xff08;4&＃xff09;Application

application是Spark API 编程的应用程序&＃xff0c;它包括实现Driver功能的代码和在程序中各个executor上要执行的代码&＃xff0c;一个application由多个job组成。其中应用程序的入口为用户所定义的main方法。

&＃xff08;5&＃xff09;Driver

驱动器节点&＃xff0c;它是一个运行Application中main函数并创建SparkContext的进程。application通过Driver 和Cluster Manager及executor进行通讯。它可以运行在application节点上&＃xff0c;也可以由application提交给Cluster Manager&＃xff0c;再由Cluster Manager安排worker进行运行。

Driver节点也负责提交Job&＃xff0c;并将Job转化为Task&＃xff0c;在各个Executor进程间协调Task的调度。

&＃xff08;6&＃xff09;sparkContext

sparkContext是整个spark应用程序最关键的一个对象&＃xff0c;是Spark所有功能的主要入口点。核心作用是初始化spark应用程序所需要的组件&＃xff0c;同时还负责向master程序进行注册等。

3. spark其它核心概念

&＃xff08;1&＃xff09;RDD

它是Spark中最重要的一个概念&＃xff0c;是弹性分布式数据集&＃xff0c;是一种容错的、可以被并行操作的元素集合&＃xff0c;是Spark对所有数据处理的一种基本抽象。可以通过一系列的算子对rdd进行操作&＃xff0c;主要分为Transformation和Action两种操作。

• ‍‍‍‍‍Transformation(转换)&＃xff1a;是对已有的RDD进行换行生成新的RDD&＃xff0c;对于转换过程采用惰性计算机制&＃xff0c;不会立即计算出结果。常用的方法有map&＃xff0c;filter&＃xff0c;flatmap等。

• Action(执行)&＃xff1a;对已有对RDD对数据执行计算产生结果&＃xff0c;并将结果返回Driver或者写入到外部存储中。常用到方法有reduce&＃xff0c;collect&＃xff0c;saveAsTextFile等。

&＃xff08;2&＃xff09;DAG

DAG是一个有向无环图&＃xff0c;在Spark中&＃xff0c; 使用 DAG 来描述我们的计算逻辑。主要分为DAG Scheduler 和Task Scheduler。

图片出自&＃xff1a;https://blog.csdn.net/newchitu/article/details/92796302

&＃xff08;3&＃xff09;DAG Scheduler

DAG Scheduler 是面向stage的高层级的调度器&＃xff0c;DAG Scheduler把DAG拆分为多个Task&＃xff0c;每组Task都是一个stage&＃xff0c;解析时是以shuffle为边界进行反向构建的&＃xff0c;每当遇见一个shuffle&＃xff0c;spark就会产生一个新的stage&＃xff0c;接着以TaskSet的形式提交给底层的调度器&＃xff08;task scheduler&＃xff09;&＃xff0c;每个stage封装成一个TaskSet。DAG Scheduler需要记录RDD被存入磁盘物化等动作&＃xff0c;同时会需要Task寻找最优等调度逻辑&＃xff0c;以及监控因shuffle跨节点输出导致的失败。

&＃xff08;4&＃xff09;Task Scheduler

Task Scheduler 负责每一个具体任务的执行。它的主要职责包括

• 任务集的调度管理&＃xff1b;

• 状态结果跟踪&＃xff1b;

• 物理资源调度管理&＃xff1b;

• 任务执行&＃xff1b;

• 获取结果。

&＃xff08;5&＃xff09;Job

job是有多个stage构建的并行的计算任务&＃xff0c;job是由spark的action操作来触发的&＃xff0c;在spark中一个job包含多个RDD以及作用在RDD的各种操作算子。

&＃xff08;6&＃xff09;stage

DAG Scheduler会把DAG切割成多个相互依赖的Stage&＃xff0c;划分Stage的一个依据是RDD间的宽窄依赖。

在对Job中的所有操作划分Stage时&＃xff0c;一般会按照倒序进行&＃xff0c;即从Action开始&＃xff0c;遇到窄依赖操作&＃xff0c;则划分到同一个执行阶段&＃xff0c;遇到宽依赖操作&＃xff0c;则划分一个新的执行阶段&＃xff0c;且新的阶段为之前阶段的parent&＃xff0c;然后依次类推递归执行。

child Stage需要等待所有的parent Stage执行完之后才可以执行&＃xff0c;这时Stage之间根据依赖关系构成了一个大粒度的DAG。在一个Stage内&＃xff0c;所有的操作以串行的Pipeline的方式&＃xff0c;由一组Task完成计算。

&＃xff08;7&＃xff09;TaskSet Task

TaskSet 可以理解为一种任务&＃xff0c;对应一个stage&＃xff0c;是Task组成的任务集。一个TaskSet中的所有Task没有shuffle依赖可以并行计算。

Task是spark中最独立的计算单元&＃xff0c;由Driver Manager发送到executer执行&＃xff0c;通常情况一个task处理spark RDD一个partition。Task分为ShuffleMapTask和ResultTask两种&＃xff0c;位于最后一个Stage的Task为ResultTask&＃xff0c;其他阶段的属于ShuffleMapTask。

四、spark作业运行流程

1. spark作业运行流程

spark应用程序以进程集合为单位在分布式集群上运行&＃xff0c;通过driver程序的main方法创建sparkContext的对象与集群进行交互。具体运行流程如下&＃xff1a;

• sparkContext向cluster Manager申请CPU&＃xff0c;内存等计算资源。

• cluster Manager分配应用程序执行所需要的资源&＃xff0c;在worker节点创建executor。

• sparkContext将程序代码和task任务发送到executor上进行执行&＃xff0c;代码可以是编译成的jar包或者python文件等。接着sparkContext会收集结果到Driver端。
  

2. spark RDD迭代过程

• sparkContext创建RDD对象&＃xff0c;计算RDD间的依赖关系&＃xff0c;并组成一个DAG有向无环图。

• DAGScheduler将DAG划分为多个stage&＃xff0c;并将stage对应的TaskSet提交到集群的管理中心&＃xff0c;stage的划分依据是RDD中的宽窄依赖&＃xff0c;spark遇见宽依赖就会划分为一个stage&＃xff0c;每个stage中包含来一个或多个task任务&＃xff0c;避免多个stage之间消息传递产生的系统开销。

• taskScheduler 通过集群管理中心为每一个task申请资源并将task提交到worker的节点上进行执行。

• worker上的executor执行具体的任务。
  

3. yarn资源管理器介绍

spark 程序一般是运行在集群上的&＃xff0c;spark on yarn是工作或生产上用的非常多的一种运行模式。

没有yarn模式前&＃xff0c;每个分布式框架都要跑在一个集群上面&＃xff0c;比如说Hadoop要跑在一个集群上&＃xff0c;Spark用集群的时候跑在standalone上。这样的话整个集群的资源的利用率低&＃xff0c;且管理起来比较麻烦。

yarn是分布式资源管理和任务管理管理&＃xff0c;主要由ResourceManager&＃xff0c;NodeManager和ApplicationMaster三个模块组成。

ResourceManager 主要负责集群的资源管理&＃xff0c;监控和分配。对于所有的应用它有绝对的控制权和资源管理权限。

NodeManager 负责节点的维护&＃xff0c;执行和监控task运行状况。会通过心跳的方式向ResourceManager汇报自己的资源使用情况。

yarn资源管理器的每个节点都运行着一个NodeManager&＃xff0c;是ResourceManager的代理。如果主节点的ResourceManager宕机后&＃xff0c;会连接ResourceManager的备用节点。

ApplicationMaster 负责具体应用程序的调度和资源的协调&＃xff0c;它会与ResourceManager协商进行资源申请。ResourceManager以container容器的形式将资源分配给application进行运行。同时负责任务的启停。

container 是资源的抽象&＃xff0c;它封装着每个节点上的资源信息(cpu,内存&＃xff0c;磁盘&＃xff0c;网络等)&＃xff0c;yarn将任务分配到container上运行&＃xff0c;同时该任务只能使用container描述的资源&＃xff0c;达到各个任务间资源的隔离。

4. spark程序在yarn上执行流程

spark on yarn分为两种模式yarn-client模式&＃xff0c;和yarn—cluster模式&＃xff0c;一般线上采用的是yarn-cluster模式。

&＃xff08;1&＃xff09;yarn-client模式

driver在客户端本地执行&＃xff0c;这种模式可以使得spark application和客户端进行交互&＃xff0c;因为driver在客户端可以通过webUI访问driver的状态。同时Driver会与yarn集群中的Executor进行大量的通信&＃xff0c;会造成客户机网卡流量的大量增加。

&＃xff08;2&＃xff09;yarn-cluster模式

Yarn-Cluster主要用于生产环境中&＃xff0c;因为Driver运行在Yarn集群中某一台NodeManager中&＃xff0c;每次提交任务的Driver所在的机器都是随机的&＃xff0c;不会产生某一台机器网卡流量激增的现象&＃xff0c;缺点是任务提交后不能看到日志。只能通过yarn查看日志。

下图是yarn-cluster运行模式&＃xff1a;

client 向yarn提交应用程序&＃xff0c;包含ApplicationMaster程序、启动ApplicationMaster的命令、需要在Executor中运行的程序等。

ApplicationMaster程序启动ApplicationMaster的命令、需要在Executor中运行的程序等。

ApplicationMaster向ResourceManager注册&＃xff0c;这样用户可以直接通过ResourceManage查看应用程序的运行状态。

ApplicationMaster申请到资源&＃xff08;也就是Container&＃xff09;后&＃xff0c;便与对应的NodeManager通信&＃xff0c;启动Task。

Task向ApplicationMaster汇报运行的状态和进度&＃xff0c;以让ApplicationMaster随时掌握各个任务的运行状态&＃xff0c;从而可以在任务失败时重新启动任务。

应用程序运行完成后&＃xff0c;ApplicationMaster向ResourceManager申请注销并关闭自己。

参考文献&＃xff1a;

[1] Spark on Yarn架构原理:

https://blog.csdn.net/lijingjingchn/article/details/85012470

[1] Spark on Yarn详解:

https://www.cnblogs.com/bigdata1024/p/12116621.html

[1] Spark任务提交方式和执行流程:

https://www.cnblogs.com/frankdeng/p/9301485.html

[1] Spark容错机制:

https://www.cnblogs.com/cynchanpin/p/7163160.html

[1] spark 之 scheduler:

https://mp.weixin.qq.com/s/9g5e5WlmXUyQDXiU6PTGZA?token&＃61;1292183487&lang&＃61;zh_CN

[1] Spark工作原理:

https://blog.csdn.net/qq_16681169/article/details/82432841

[1] Spark RDD:

https://www.cnblogs.com/zlslch/p/5942204.html

[1] Spark基本概念快速入门:

https://www.leonlu.cc/profession/17-spark-terminology/

[1] spark中dag的介绍:

https://blog.csdn.net/newchitu/article/details/92796302

[2] spark:

https://spark.apache.org/docs/3.0.0-preview/index.html