1、spark是什么?
快速,通用,可扩展的分布式计算引擎
2、弹性分布式数据集RDD
RDD(Resilient Distributed Dataset)叫做分布式数据集,是Spark中最基本的数据抽象,它代表一个不可变、可分区、里面的元素可并行计算的集合。RDD具有数据流模型的特点:自动容错、位置感知性调度和可伸缩性。RDD允许用户在执行多个查询时显式地将工作集缓存在内存中,后续的查询能够重用工作集,这极大地提升了查询速度。
RDD的属性
1)一组分片(Partition),即数据集的基本组成单位。对于RDD来说,每个分片都会被一个计算任务处理,并决定并行计算的粒度。用户可以在创建RDD时指定RDD的分片个数,如果没有指定,那么就会采用默认值。默认值就是程序所分配到的CPU Core的数目。
2)一个计算每个分区的函数。Spark中RDD的计算是以分片为单位的,每个RDD都会实现compute函数以达到这个目的。compute函数会对迭代器进行复合,不需要保存每次计算的结果。
3)RDD之间的依赖关系。RDD的每次转换都会生成一个新的RDD,所以RDD之间就会形成类似于流水线一样的前后依赖关系。在部分分区数据丢失时,Spark可以通过这个依赖关系重新计算丢失的分区数据,而不是对RDD的所有分区进行重新计算。
4)一个Partitioner,即RDD的分片函数。当前Spark中实现了两种类型的分片函数,一个是基于哈希的HashPartitioner,另外一个是基于范围的RangePartitioner。只有对于于key-value的RDD,才会有Partitioner,非key-value的RDD的Parititioner的值是None。Partitioner函数不但决定了RDD本身的分片数量,也决定了parent RDD Shuffle输出时的分片数量。
5)一个列表,存储存取每个Partition的优先位置(preferred location)。对于一个HDFS文件来说,这个列表保存的就是每个Partition所在的块的位置。按照“移动数据不如移动计算”的理念,Spark在进行任务调度的时候,会尽可能地将计算任务分配到其所要处理数据块的存储位置。
创建RDD的两种方式
1、由一个已经存在的Scala集合创建。
val rdd1 = sc.parallelize(Array(1,2,3,4,5,6,7,8))
2、由外部存储系统的数据集创建,包括本地的文件系统,还有所有Hadoop支持的数据集,比如HDFS、Cassandra、HBase等
val rdd2 = sc.textFile("hdfs://node1.itcast.cn:9000/words.txt")
3、Spark的算子
RDD中的所有转换都是延迟加载的,也就是说,它们并不会直接计算结果。相反的,它们只是记住这些应用到基础数据集(例如一个文件)上的转换动作。只有当发生一个要求返回结果给Driver的动作时,这些转换才会真正运行。这种设计让Spark更加有效率地运行。
1、Transformation
2、Action
4、RDD的依赖关系
RDD和它依赖的父RDD(s)的关系有两种不同的类型,即窄依赖(narrow dependency)和宽依赖(wide dependency)。
RDD缓存
Spark速度非常快的原因之一,就是在不同操作中可以在内存中持久化或缓存个数据集。当持久化某个RDD后,每一个节点都将把计算的分片结果保存在内存中
RDD缓存的方式
RDD通过persist方法或cache方法可以将前面的计算结果缓存,但是并不是这两个方法被调用时立即缓存,而是触发后面的action时,该RDD将会被缓存在计算节点的内存中,并供后面重用。
cache最终也是调用了persist方法,默认的存储级别都是仅在内存存储一份
Spark的存储级别
缓存有可能丢失,或者存储存储于内存的数据由于内存不足而被删除,RDD的缓存容错机制保证了即使缓存丢失也能保证计算的正确执行。通过基于RDD的一系列转换,丢失的数据会被重算,由于RDD的各个Partition是相对独立的,因此只需要计算丢失的部分即可,并不需要重算全部Partition。
5、DAG的生成
DAG(Directed Acyclic Graph)叫做有向无环图,原始的RDD通过一系列的转换就就形成了DAG,根据RDD之间的依赖关系的不同将DAG划分成不同的Stage,对于窄依赖,partition的转换处理在Stage中完成计算。对于宽依赖,由于有Shuffle的存在,只能在parent RDD处理完成后,才能开始接下来的计算,因此宽依赖是划分Stage的依据。
spark运行原理
1、通过ActorSystem创建MasterActor,启动定时器,定时检查与接收Worker节点的发送消息
2、Worker节点主动向Master发送注册消息
3、Master接收Worker的注册请求,然后将注册信息保存起来,并向Worker返回一个注册成功的消息
4、Worker接收到Master注册成功的消息后,启用定时器,定时向master发送心跳报活,Master接收到Worker发送来的心跳消息后,更新Worker上一次的心跳时间
5、DAGScheduler根据FinalRDD递归向上解析Lineager的依赖关系,并以宽依赖为切分一个新stage的依据,并将多个task任务封装到TaskSet,其中Task的数量由其父RDD的切片数量决定,最后使用递归优先提交父Stage(TaskSet)
6、先创建TaskScheduler即TaskSchedulerImpl接着又创建SparkDeploySchedulerBackend对资源参数创建AppClient与Master注册Application,并替每个TaskSet创建TaskManager负责监控此TaskSet中任务的执行情况
7、Master接收到ClientActor的任务描述之后,将任务描述信息保存起来,然后ClientActor返回消息,告知ClientActor任务注册成功,接下来Master(打散|负载均衡|尽量集中)进行资源调度
8、Master跟Worker通信,然后让Worker启动Executor
9、Executor向Driver发送注册消息,Driver接收到Executor注册消息后,响应注册成功的消息
10、Executor接收到Driver注册成功的消息后,本进程中创建Executor的引用对象
11、Driver中TaskSchedulerImp向Executor发送LaunchTask消息,Executor将创建一个线程池作为所提交的Task任务的容器
12、Task接收到launchTask消息后,准备运行文件初始化与反序列化,就绪后,调用Task的run方法,其中每个Task所执行的函数是应用在RDD中的一个独立分区上
13、Task运行完成,向TaskManager汇报情况,并且释放线程资源
14、所有Task运行结束之后,Executor向Worker注销自身,释放资源。
Spark Standalone
Spark Standalone模式中,资源调度室Spark自行实现的,其节点类型分为master和worker,
其中Driver运行在Master中,并且有长驻内存的Master进程守护,Worker节点上常驻Worker守护进程,负责与Master节点通信,通过ExecutorRunner来控制运行在当前节点上的CoarseGrainedExecutorBackend,
每个Worker上存在一个或多个CoarseGrainedExecutorBackend进程,每个进程包含一个Executor对象,该对象持有一个线程池,每个线程池可以执行一个Task
节点类型:
1. master 节点: 常驻master进程,负责管理全部worker节点。
2. worker 节点: 常驻worker进程,负责管理executor 并与master节点通信。
dirvier:官方解释为: The process running the main() function of the application and creating the SparkContext。即理解为用户自己编写的应用程序
Executor:执行器:
在每个WorkerNode上为某应用启动的一个进程,该进程负责运行任务,并且负责将数据存在内存或者磁盘上,每个任务都有各自独立的Executor。
Executor是一个执行Task的容器。它的主要职责是:
1、初始化程序要执行的上下文SparkEnv,解决应用程序需要运行时的jar包的依赖,加载类。
2、同时还有一个ExecutorBackend向cluster manager汇报当前的任务状态,这一方面有点类似hadoop的tasktracker和task。
总结:Executor是一个应用程序运行的监控和执行容器。Executor的数目可以在submit时,由 --num-executors (on yarn)指定.
Job:
包含很多task的并行计算,可以认为是Spark RDD 里面的action,每个action的计算会生成一个job。
用户提交的Job会提交给DAGScheduler,Job会被分解成Stage和Task。
Stage:
一个Job会被拆分为多组Task,每组任务被称为一个Stage就像Map Stage, Reduce Stage。
Stage的划分在RDD的论文中有详细的介绍,简单的说是以shuffle和result这两种类型来划分。在Spark中有两类task,一类是shuffleMapTask,一类是resultTask,第一类task的输出是shuffle所需数据,第二类task的输出是result,stage的划分也以此为依据,shuffle之前的所有变换是一个stage,shuffle之后的操作是另一个stage。比如 rdd.parallize(1 to 10).foreach(println) 这个操作没有shuffle,直接就输出了,那么只有它的task是resultTask,stage也只有一个;如果是rdd.map(x => (x, 1)).reduceByKey(_ + _).foreach(println), 这个job因为有reduce,所以有一个shuffle过程,那么reduceByKey之前的是一个stage,执行shuffleMapTask,输出shuffle所需的数据,reduceByKey到最后是一个stage,直接就输出结果了。如果job中有多次shuffle,那么每个shuffle之前都是一个stage。
Task
即 stage 下的一个任务执行单元,一般来说,一个 rdd 有多少个 partition,就会有多少个 task,因为每一个 task 只是处理一个 partition 上的数据.
每个executor执行的task的数目, 可以由submit时,--num-executors(on yarn) 来指定。
小结:
驱动程序就是执行了一个Spark Application的main函数和创建Spark Context的进程,它包含了这个application的全部代码。
Spark Application中的每个action会被Spark作为Job进行调度。
每个Job是一个计算序列的最终结果,而这个序列中能够产生中间结果的计算就是一个stage。
对于Transformations和Actions是有着明确区分的。通常Action对应了Job,而Transformation对应了Stage
一个Job被拆分成若干个Stage,每个Stage执行一些计算,产生一些中间结果。它们的目的是最终生成这个Job的计算结果。
而每个Stage是一个task set,包含若干个task。Task是Spark中最小的工作单元,在一个executor上完成一个特定的事情。
1、driver program是用户写的带main函数的代码
2、每个action算子的操作都会对应一个job,例如(ForeachRDD写入外部系统的一个操作)
3、DAGScheduler会对Job进行拆分,拆分的依据:根据FinalRDD(在这里ForeachRDD)递归向上解析Lineager的依赖关系,以宽依赖为切分stage的依据,切分成若干个Stage,递归优先提交父Stage,每个Stage里面包含多个Task任务
4、若干个Transformation的算子RDD组成Stage,所以一个RDD中有多少个partition,就有多少个Task,因为每一个Task只对一个partition数据做处理。