Spark架构与作业执行流程简介(scala版)

  在讲spark之前,不得不详细介绍一下RDD(Resilient Distributed Dataset),打开RDD的源码,一开始的介绍如此:

Spark架构与作业执行流程简介(scala版)

字面意思就是弹性分布式数据集,是spark中最基本的数据抽象,它代表一个不可变、可分区、里面的元素可并行计算的集合。

Resilient:弹性的,它表示的是数据可以保存在磁盘,也可以保存在内存中

Distributed:它的数据分布式存储,并且可以做分布式的计算

Dataset:一个数据集,简单的理解为集合,用于存放数据的

事实上,关于RDD有5个特性,同样我们可以看看源码是怎么介绍这5个特性的:

Spark架构与作业执行流程简介(scala版)

  1. A list of partitions (每个RDD都有一个分区列表)
  2. A function for computing each split (作用在每个分区上面的函数)
  3. A list of dependencies on other RDDs (一个RDD依赖其他多个RDD,这个特性很重要,rdd的容错机制就是根据这个特性而来的)
  4. Optionally, a Partitioner for key-value RDDs (e.g. to say that the RDD is hash-partitioned) [可选项:针对于 kv 键值对的RDD才具有该分区特性]
  5. Optionally, a list of preferred locations to compute each split on (e.g. block locations foran HDFS file) [可选项 : 数据本地性,数据最优,选择尽量存储在worker节点上的数据节点。]

那么我们又如何更好的去理解这5个特性呢?

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)    传统的MapReduce虽然具有自动容错、平衡负载和可拓展性的优点,但是其最大缺点是采用非循环式的数据流模型,使得在迭代计算式要进行大量的磁盘IO操作。RDD正是解决这一缺点的抽象方法

(2)     RDD是Spark提供的最重要的抽象的概念,它是一种有容错机制的特殊集合,可以分布在集群的节点上,以函数式编操作集合的方式,进行各种并行操作。可以将RDD理 解为一个具有容错机制的特殊集合,它提供了一种只读、只能有已存在的RDD变换而来的共享内存,然后将所有数据都加载到内存中,方便进行多次重用。

    a. 他是分布式的,可以分布在多台机器上,进行计算。

b. 他是弹性的,计算过程中内存不够时它会和磁盘进行数据交换。

c. 这些限制可以极大的降低自动容错开销

d. 实质是一种更为通用的迭代并行计算框架,用户可以显示的控制计算的中间结果,然后将其*运用于之后的计算。

(3)    RDD的容错机制实现分布式数据集容错方法有两种:数据检查点和记录更新RDD

  采用记录更新的方式:记录所有更新点的成本很高。所以,RDD只支持粗  颗粒变换,即只记录单个块上执行的单个操作,然后创建某个RDD的变换序列(血统)存储下来;变换序列指,每个RDD都包含了他是如何由其他RDD变换过来的以及如何重建某一块数据的信息。因此RDD的容错机制又称“血统”容错。 要实现这种“血统”容错机制,最大的难题就是如何表达父RDD和子RDD之间的依赖关系。实际上依赖关系可以分两种,窄依赖和宽依赖:窄依赖:子RDD中的每个数据块只依赖于父RDD中对应的有限个固定的数据块;宽依赖:子RDD中的一个数据块可以依赖于父RDD中的所有数据块。

(4)RDD内部的设计每个RDD都需要包含以下四个部分:

    a. 源数据分割后的数据块,源代码中的splits变量

b. 关于“血统”的信息,源码中的dependencies变量

c. 一个计算函数(该RDD如何通过父RDD计算得到),源码中的iterator(split)和compute函数

d. 一些关于如何分块和数据存放位置的元信息,如源码中的partitioner和preferredLocations

思考:RDD在spark中的地位及作用

(1)    为什么会有Spark?

因为传统的并行计算模型无法有效的解决迭代计算(iterative)和交互式计算(interactive);而Spark的使命便是解决这两个问题,这也是他存在的价值和理由。

(2)    Spark如何解决迭代计算?

其主要实现思想就是RDD,把所有计算的数据保存在分布式的内存中。迭代计算通常情况下都是对同一个数据集做反复的迭代计算,数据在内存中将大大提升IO操作。这也是Spark涉及的核心:内存计算。

(3)    Spark如何实现交互式计算?

因为Spark是用scala语言实现的,Spark和scala能够紧密的集成,所以Spark可以完美的运用scala的解释器,使得其中的scala可以向操作本地集合对象一样轻松操作分布式数据集。

(4)    Spark和RDD的关系?

可以理解为:RDD是一种具有容错性基于内存的集群计算抽象方法,RDD是Spark Core的底层核心,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("/words.txt")

3)通过已有的RDD经过算子转换生成新的RDD

val rdd3=rdd2.flatMap(_.split(" "))

RDD编程API之RDD的算子分类

RDD的算子分类可以分为2种,TransformationAction类 。

Transformation:根据数据集创建一个新的数据集,计算后返回一个新RDD;例如:Map将数据的每个元素经过某个函数计算后,返回一个新的分布式数据集。

Action:对数据集计算后返回一个数值value给驱动程序;例如:collect将数据集的所有元素收集完成返回给程序。

关于Transformation:

RDD中的所有转换都是延迟加载的,也就是说,它们并不会直接计算结果。相反的,它们只是记住这些应用到基础数据集(例如一个文件)上的转换动作。只有当发生一个要求返回结果给Driver的动作时,这些转换才会真正运行。这种设计让Spark更加有效率地运行。

常用的Transformation:

转换

含义

map(func)

返回一个新的RDD,该RDD由每一个输入元素经过func函数转换后组成

filter(func)

返回一个新的RDD,该RDD由经过func函数计算后返回值为true的输入元素组成

flatMap(func)

类似于map,但是每一个输入元素可以被映射为0或多个输出元素(所以func应该返回一个序列,而不是单一元素)

mapPartitions(func)

类似于map,但独立地在RDD的每一个分片上运行,因此在类型为T的RDD上运行时,func的函数类型必须是Iterator[T] => Iterator[U]

mapPartitionsWithIndex(func)

类似于mapPartitions,但func带有一个整数参数表示分片的索引值,因此在类型为T的RDD上运行时,func的函数类型必须是

(Int, Interator[T]) => Iterator[U]

sample(withReplacement, fraction, seed)

根据fraction指定的比例对数据进行采样,可以选择是否使用随机数进行替换,seed用于指定随机数生成器种子

union(otherDataset)

对源RDD和参数RDD求并集后返回一个新的RDD

intersection(otherDataset)

对源RDD和参数RDD求交集后返回一个新的RDD

distinct([numTasks]))

对源RDD进行去重后返回一个新的RDD

groupByKey([numTasks])

在一个(K,V)的RDD上调用,返回一个(K, Iterator[V])的RDD

reduceByKey(func, [numTasks])

在一个(K,V)的RDD上调用,返回一个(K,V)的RDD,使用指定的reduce函数,将相同key的值聚合到一起,与groupByKey类似,reduce任务的个数可以通过第二个可选的参数来设置

aggregateByKey(zeroValue)(seqOp, combOp, [numTasks])

sortByKey([ascending], [numTasks])

在一个(K,V)的RDD上调用,K必须实现Ordered接口,返回一个按照key进行排序的(K,V)的RDD

sortBy(func,[ascending], [numTasks])

与sortByKey类似,但是更灵活

join(otherDataset, [numTasks])

在类型为(K,V)和(K,W)的RDD上调用,返回一个相同key对应的所有元素对在一起的(K,(V,W))的RDD

cogroup(otherDataset, [numTasks])

在类型为(K,V)和(K,W)的RDD上调用,返回一个(K,(Iterable<V>,Iterable<W>))类型的RDD

cartesian(otherDataset)

笛卡尔积

pipe(command, [envVars])

对rdd进行管道操作

coalesce(numPartitions)

减少 RDD 的分区数到指定值。在过滤大量数据之后,可以执行此操作

repartition(numPartitions)

重新给 RDD 分区

repartitionAndSortWithinPartitions(partitioner)

重新给 RDD 分区,并且每个分区内以记录的 key 排序

常用的Action:

动作

含义

reduce(func)

通过func函数聚集RDD中的所有元素,这个功能必须是课交换且可并联的

collect()

在驱动程序中,以数组的形式返回数据集的所有元素

count()

返回RDD的元素个数

first()

返回RDD的第一个元素(类似于take(1))

take(n)

返回一个由数据集的前n个元素组成的数组

takeSample(withReplacement,num, [seed])

返回一个数组,该数组由从数据集中随机采样的num个元素组成,可以选择是否用随机数替换不足的部分,seed用于指定随机数生成器种子

takeOrdered(n, [ordering])

返回自然顺序或者自定义顺序的前 n 个元素

saveAsTextFile(path)

将数据集的元素以textfile的形式保存到HDFS文件系统或者其他支持的文件系统,对于每个元素,Spark将会调用toString方法,将它装换为文件中的文本

saveAsSequenceFile(path) 

将数据集中的元素以Hadoop sequencefile的格式保存到指定的目录下,可以使HDFS或者其他Hadoop支持的文件系统。

saveAsObjectFile(path) 

将数据集的元素,以 Java 序列化的方式保存到指定的目录下

countByKey()

针对(K,V)类型的RDD,返回一个(K,Int)的map,表示每一个key对应的元素个数。

foreach(func)

在数据集的每一个元素上,运行函数func进行更新。

RDD的依赖关系:

RDD和它依赖的父RDD(s)的关系有两种不同的类型,即窄依赖(narrow dependency)和宽依赖(wide dependency)。

Spark架构与作业执行流程简介(scala版)

窄依赖:窄依赖指的是每一个父RDD的Partition最多被子RDD的一个Partition使用

总结:窄依赖我们形象的比喻为独生子女

宽依赖:宽依赖指的是多个子RDD的Partition会依赖同一个父RDD的Partition

总结:宽依赖我们形象的比喻为超生

Lineage血统:

RDD只支持粗粒度转换,即在大量记录上执行的单个操作。将创建RDD的一系列Lineage(即血统)记录下来,以便恢复丢失的分区。RDD的Lineage会记录RDD的元数据信息和转换行为,当该RDD的部分分区数据丢失时,它可以根据这些信息来重新运算和恢复丢失的数据分区。

RDD的缓存:

Spark速度非常快的原因之一,就是在不同操作中可以在内存中持久化或缓存个数据集。当持久化某个RDD后,每一个节点都将把计算的分片结果保存在内存中,并在对此RDD或衍生出的RDD进行的其他动作中重用。这使得后续的动作变得更加迅速。RDD相关的持久化和缓存,是Spark最重要的特征之一。可以说,缓存是Spark构建迭代式算法和快速交互式查询的关键。  

RDD缓存方式:

RDD通过persist方法或cache方法可以将前面的计算结果缓存,但是并不是这两个方法被调用时立即缓存,而是触发后面的action时,该RDD将会被缓存在计算节点的内存中,并供后面重用。

Spark架构与作业执行流程简介(scala版)

通过查看源码发现cache最终也是调用了persist方法,默认的存储级别都是仅在内存存储一份,Spark的存储级别还有好多种,存储级别在object StorageLevel中定义的。

Spark架构与作业执行流程简介(scala版)

缓存有可能丢失,或者存储于内存的数据由于内存不足而被删除,RDD的缓存容错机制保证了即使缓存丢失也能保证计算的正确执行。通过基于RDD的一系列转换,丢失的数据会被重算,由于RDD的各个Partition是相对独立的,因此只需要计算丢失的部分即可,并不需要重算全部Partition。

什么是DAG:

DAG(Directed Acyclic Graph)叫做有向无环图,原始的RDD通过一系列的转换就就形成了DAG,根据RDD之间的依赖关系的不同将DAG划分成不同的Stage,对于窄依赖,partition的转换处理在Stage中完成计算。对于宽依赖,由于有Shuffle的存在,只能在parent RDD处理完成后,才能开始接下来的计算,因此宽依赖是划分Stage的依据。

Spark架构与作业执行流程简介(scala版)

spark的任务调度流程图:

Spark架构与作业执行流程简介(scala版)

各个RDD之间存在着依赖关系,这些依赖关系形成有向无环图DAG,DAGScheduler对这些依赖关系形成的DAG进行Stage划分,划分的规则很简单,从后往前回溯,遇到窄依赖加入本stage,遇见宽依赖进行Stage切分。完成了Stage的划分,DAGScheduler基于每个Stage生成TaskSet,并将TaskSet提交给TaskScheduler。TaskScheduler 负责具体的task调度,最后在Worker节点上启动task。

DAGScheduler的功能 :

(1)DAGScheduler构建Stage

(2)记录哪个RDD或者 Stage 输出被物化(缓存),通常在一个复杂的shuffle之后,通常物化一下(cache、persist),方便之后的计算。

(3)重新提交shuffle输出丢失的stage(stage内部计算出错)

(4)将 Taskset 传给底层调度器

   a)– spark-cluster TaskScheduler

   b)– yarn-cluster YarnClusterScheduler

   c)– yarn-client YarnClientClusterScheduler

TaskScheduler的 功能:

(1)为每一个TaskSet构建一个TaskSetManager 实例管理这个TaskSet 的生命周期

(2)数据本地性决定每个Task最佳位置

(3)提交 taskset( 一组task) 到集群运行并监控

(4)推测执行,碰到 straggle(计算缓慢)任务需要放到别的节点上重试

(5)重新提交Shuffle输出丢失的Stage给DAGScheduler

RDD容错机制之checkpoint

checkpoint是什么:

(1)Spark 在生产环境下经常会面临transformation的RDD非常多(例如一个Job中包含1万个RDD)或者具体transformation的RDD本身计算特别复杂或者耗时(例如计算时长超过1个小时),这个时候就要考虑对计算结果数据的持久化;

(2)Spark是擅长多步骤迭代的,同时擅长基于Job的复用,这个时候如果能够对曾经计算的过程产生的数据进行复用,就可以极大的提升效率;

(3)如果采用persist把数据放在内存中,虽然是快速的,但是也是最不可靠的;如果把数据放在磁盘上,也不是完全可靠的!例如磁盘会损坏,系统管理员可能清空磁盘

(4)Checkpoint的产生就是为了相对而言更加可靠的持久化数据,在Checkpoint的时候可以指定把数据放在本地,并且是多副本的方式,但是在生产环境下是放在HDFS上,这就天然的借助了HDFS高容错、高可靠的特征来完成了最大化的可靠的持久化数据的方式;

(5)Checkpoint是为了最大程度保证绝对可靠的复用RDD计算数据的Spark高级功能,通过checkpoint我们通常把数据持久化到HDFS来保证数据最大程度的安全性;

(6)Checkpoint就是针对整个RDD计算链条中特别需要数据持久化的环节(后面会反复使用当前环节的RDD)开始基于HDFS等的数据持久化复用策略,通过对RDD启动checkpoint机制来实现容错和高可用;由此当加入进行一个1万个步骤,在9000个步骤的时候persist,数据还是有可能丢失的,但是如果checkpoint,数据丢失的概率几乎为0。

checkpoint原理机制:

(1)当RDD使用cache机制从内存中读取数据,如果数据没有读到,会使用checkpoint机制读取数据。此时如果没有checkpoint机制,那么就需要找到父RDD重新计算数据了,因此checkpoint是个很重要的容错机制。checkpoint就是对于一个RDD chain(链),如果中间某些中间结果RDD,后面需要反复使用该数据,可能因为一些故障导致该中间数据丢失,那么就可以针对该RDD启动checkpoint机制,checkpoint,首先需要调用sparkContext的setCheckpoint方法,设置一个容错文件系统目录,比如hdfs,然后对RDD调用checkpoint方法。之后再RDD所处的job运行结束后,会启动一个单独的job,来将checkpoint过的数据写入之前设置的文件系统持久化,进行高可用。所以后面的计算在使用该RDD时,如果数据丢失了,但是还是可以从它的checkpoint中读取数据,不需要重新计算。

(2)persist或者cache与checkpoint的区别在于,前者持久化只是将数据保存在BlockManager中但是其lineage是不变的,但是后者checkpoint执行完后,rdd已经没有依赖RDD,只有一个checkpointRDD,checkpoint之后,RDD的lineage就改变了。而且,持久化的数据丢失的可能性更大,因为可能磁盘或内存被清理,但是checkpoint的数据通常保存到hdfs上,放在了高容错文件系统。

Spark运行架构

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

Spark架构与作业执行流程简介(scala版)

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运行架构特点:

  • 每个Application获取专属的executor进程,该进程在Application期间一直驻留,并以多线程方式运行tasks。这种Application隔离机制有其优势的,无论是从调度角度看(每个Driver调度它自己的任务),还是从运行角度看(来自不同Application的Task运行在不同的JVM中)。当然,这也意味着Spark Application不能跨应用程序共享数据,除非将数据写入到外部存储系统。
  • Spark与资源管理器无关,只要能够获取executor进程,并能保持相互通信就可以了。

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

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

上一篇:linux之无名管道


下一篇:轻松学JVM(二)——内存模型、可见性、指令重排序