怎样进行Spark Core的详细分析

先来一个问题，也是面试中常问的：

Spark为什么会流行？

原因1：优秀的数据模型和丰富计算抽象

Spark 产生之前，已经有MapReduce这类非常成熟的计算系统存在了，并提供了高层次的API(map/reduce)，把计算运行在集群中并提供容错能力，从而实现分布式计算。

虽然MapReduce提供了对数据访问和计算的抽象，但是对于数据的复用就是简单的将中间数据写到一个稳定的文件系统中(例如HDFS)，所以会产生数据的复制备份，磁盘的I/O以及数据的序列化，所以在遇到需要在多个计算之间复用中间结果的操作时效率就会非常的低。而这类操作是非常常见的，例如迭代式计算，交互式数据挖掘，图计算等。

认识到这个问题后，学术界的 AMPLab 提出了一个新的模型，叫做 RDD。RDD 是一个可以容错且并行的数据结构(其实可以理解成分布式的集合，操作起来和操作本地集合一样简单)，它可以让用户显式的将中间结果数据集保存在内存中，并且通过控制数据集的分区来达到数据存放处理最优化。同时 RDD也提供了丰富的 API (map、reduce、filter、foreach、redeceByKey...)来操作数据集。后来RDD被 AMPLab 在一个叫做 Spark 的框架中提供并开源。

简而言之，Spark 借鉴了 MapReduce 思想发展而来，保留了其分布式并行计算的优点并改进了其明显的缺陷。让中间数据存储在内存中提高了运行速度、并提供丰富的操作数据的API提高了开发速度。

原因2：完善的生态圈-fullstack

目前，Spark已经发展成为一个包含多个子项目的集合，其中包含SparkSQL、Spark Streaming、GraphX、MLlib等子项目。

Spark Core：实现了 Spark 的基本功能，包含RDD、任务调度、内存管理、错误恢复、与存储系统交互等模块。

Spark SQL：Spark 用来操作结构化数据的程序包。通过 Spark SQL，我们可以使用 SQL操作数据。

Spark Streaming：Spark 提供的对实时数据进行流式计算的组件。提供了用来操作数据流的 API。

Spark MLlib：提供常见的机器学习(ML)功能的程序库。包括分类、回归、聚类、协同过滤等，还提供了模型评估、数据导入等额外的支持功能。

GraphX(图计算)：Spark中用于图计算的API，性能良好，拥有丰富的功能和运算符，能在海量数据上自如地运行复杂的图算法。

集群管理器：Spark 设计为可以高效地在一个计算节点到数千个计算节点之间伸缩计算。

StructuredStreaming：处理结构化流,统一了离线和实时的API。

Spark VS Hadoop

❣️注意：
尽管Spark相对于Hadoop而言具有较大优势，但Spark并不能完全替代Hadoop，Spark主要用于替代Hadoop中的MapReduce计算模型。存储依然可以使用HDFS，但是中间结果可以存放在内存中；调度可以使用Spark内置的，也可以使用更成熟的调度系统YARN等。
实际上，Spark已经很好地融入了Hadoop生态圈，并成为其中的重要一员，它可以借助于YARN实现资源调度管理，借助于HDFS实现分布式存储。
此外，Hadoop可以使用廉价的、异构的机器来做分布式存储与计算，但是，Spark对硬件的要求稍高一些，对内存与CPU有一定的要求。

Spark Core

一、RDD详解

1. 为什么要有RDD?

在许多迭代式算法(比如机器学习、图算法等)和交互式数据挖掘中，不同计算阶段之间会重用中间结果，即一个阶段的输出结果会作为下一个阶段的输入。但是，之前的MapReduce框架采用非循环式的数据流模型，把中间结果写入到HDFS中，带来了大量的数据复制、磁盘IO和序列化开销。且这些框架只能支持一些特定的计算模式(map/reduce)，并没有提供一种通用的数据抽象。

AMP实验室发表的一篇关于RDD的论文:《Resilient Distributed Datasets: A Fault-Tolerant Abstraction for In-Memory Cluster Computing》就是为了解决这些问题的。

RDD提供了一个抽象的数据模型，让我们不必担心底层数据的分布式特性，只需将具体的应用逻辑表达为一系列转换操作(函数)，不同RDD之间的转换操作之间还可以形成依赖关系，进而实现管道化，从而避免了中间结果的存储，大大降低了数据复制、磁盘IO和序列化开销，并且还提供了更多的API(map/reduec/filter/groupBy...)。

2. RDD是什么?

RDD(Resilient Distributed Dataset)叫做弹性分布式数据集，是Spark中最基本的数据抽象，代表一个不可变、可分区、里面的元素可并行计算的集合。单词拆解：

• Resilient ：它是弹性的，RDD里面的中的数据可以保存在内存中或者磁盘里面
• Distributed ：它里面的元素是分布式存储的，可以用于分布式计算
• Dataset: 它是一个集合，可以存放很多元素

3. RDD主要属性

进入RDD的源码中看下：

 

在源码中可以看到有对RDD介绍的注释，我们来翻译下：

1. A list of partitions ：一组分片(Partition)/一个分区(Partition)列表，即数据集的基本组成单位。对于RDD来说，每个分片都会被一个计算任务处理，分片数决定并行度。用户可以在创建RDD时指定RDD的分片个数，如果没有指定，那么就会采用默认值。

2. A function for computing each split ：一个函数会被作用在每一个分区。Spark中RDD的计算是以分片为单位的，compute函数会被作用到每个分区上。

3. A list of dependencies on other RDDs ：一个RDD会依赖于其他多个RDD。RDD的每次转换都会生成一个新的RDD，所以RDD之间就会形成类似于流水线一样的前后依赖关系。在部分分区数据丢失时，Spark可以通过这个依赖关系重新计算丢失的分区数据，而不是对RDD的所有分区进行重新计算。(Spark的容错机制)

4. Optionally, a Partitioner for key-value RDDs (e.g. to say that the RDD is hash-partitioned)：可选项，对于KV类型的RDD会有一个Partitioner，即RDD的分区函数，默认为HashPartitioner。

5. Optionally, a list of preferred locations to compute each split on (e.g. block locations for an HDFS file)：可选项,一个列表，存储存取每个Partition的优先位置(preferred location)。对于一个HDFS文件来说，这个列表保存的就是每个Partition所在的块的位置。按照"移动数据不如移动计算"的理念，Spark在进行任务调度的时候，会尽可能选择那些存有数据的worker节点来进行任务计算。

总结

RDD 是一个数据集的表示，不仅表示了数据集，还表示了这个数据集从哪来，如何计算，主要属性包括：

1. 分区列表
2. 计算函数
3. 依赖关系
4. 分区函数(默认是hash)
5. 最佳位置

分区列表、分区函数、最佳位置，这三个属性其实说的就是数据集在哪，在哪计算更合适，如何分区；
计算函数、依赖关系，这两个属性其实说的是数据集怎么来的。

二、RDD-API

1. RDD的创建方式

1. 由外部存储系统的数据集创建，包括本地的文件系统，还有所有Hadoop支持的数据集，比如HDFS、Cassandra、HBase等：
   val rdd1 = sc.textFile("hdfs://node1:8020/wordcount/input/words.txt")

2. 通过已有的RDD经过算子转换生成新的RDD：
   val rdd2=rdd1.flatMap(_.split(" "))

3. 由一个已经存在的Scala集合创建：
   val rdd3 = sc.parallelize(Array(1,2,3,4,5,6,7,8))或
   val rdd4 = sc.makeRDD(List(1,2,3,4,5,6,7,8))

makeRDD方法底层调用了parallelize方法：

 

2. RDD的算子分类

RDD的算子分为两类:

1. Transformation转换操作:     返回一个新的RDD
2. Action动作操作:     返回值不是RDD(无返回值或返回其他的)

❣️注意:
1、RDD不实际存储真正要计算的数据，而是记录了数据的位置在哪里，数据的转换关系(调用了什么方法，传入什么函数)。
2、RDD中的所有转换都是惰性求值/延迟执行的，也就是说并不会直接计算。只有当发生一个要求返回结果给Driver的Action动作时，这些转换才会真正运行。
3、之所以使用惰性求值/延迟执行，是因为这样可以在Action时对RDD操作形成DAG有向无环图进行Stage的划分和并行优化，这种设计让Spark更加有效率地运行。

3. Transformation转换算子

4. Action动作算子

统计操作：

三、RDD的持久化/缓存

在实际开发中某些RDD的计算或转换可能会比较耗费时间，如果这些RDD后续还会频繁的被使用到，那么可以将这些RDD进行持久化/缓存，这样下次再使用到的时候就不用再重新计算了，提高了程序运行的效率。

val rdd1 = sc.textFile("hdfs://node01:8020/words.txt")
val rdd2 = rdd1.flatMap(x=>x.split(" ")).map((_,1)).reduceByKey(_+_)
rdd2.cache //缓存/持久化
rdd2.sortBy(_._2,false).collect//触发action,会去读取HDFS的文件,rdd2会真正执行持久化
rdd2.sortBy(_._2,false).collect//触发action,会去读缓存中的数据,执行速度会比之前快,因为rdd2已经持久化到内存中了

持久化/缓存API详解

• ersist方法和cache方法

RDD通过persist或cache方法可以将前面的计算结果缓存，但是并不是这两个方法被调用时立即缓存，而是触发后面的action时，该RDD将会被缓存在计算节点的内存中，并供后面重用。
通过查看RDD的源码发现cache最终也是调用了persist无参方法(默认存储只存在内存中)：

 

• 存储级别

默认的存储级别都是仅在内存存储一份，Spark的存储级别还有好多种，存储级别在object StorageLevel中定义的。

总结：

1. RDD持久化/缓存的目的是为了提高后续操作的速度
2. 缓存的级别有很多，默认只存在内存中,开发中使用memory_and_disk
3. 只有执行action操作的时候才会真正将RDD数据进行持久化/缓存
4. 实际开发中如果某一个RDD后续会被频繁的使用，可以将该RDD进行持久化/缓存

四、RDD容错机制Checkpoint

• 持久化的局限：

持久化/缓存可以把数据放在内存中，虽然是快速的，但是也是最不可靠的；也可以把数据放在磁盘上，也不是完全可靠的！例如磁盘会损坏等。

• 问题解决：

Checkpoint的产生就是为了更加可靠的数据持久化，在Checkpoint的时候一般把数据放在在HDFS上，这就天然的借助了HDFS天生的高容错、高可靠来实现数据最大程度上的安全，实现了RDD的容错和高可用。

用法：

SparkContext.setCheckpointDir("目录") //HDFS的目录

RDD.checkpoint

• 总结：

• 开发中如何保证数据的安全性性及读取效率：可以对频繁使用且重要的数据，先做缓存/持久化，再做checkpint操作。

• 持久化和Checkpoint的区别：

1. 位置：Persist 和 Cache 只能保存在本地的磁盘和内存中(或者堆外内存--实验中) Checkpoint 可以保存数据到 HDFS 这类可靠的存储上。

2. 生命周期：Cache和Persist的RDD会在程序结束后会被清除或者手动调用unpersist方法 Checkpoint的RDD在程序结束后依然存在，不会被删除。

五、RDD依赖关系

1. 宽窄依赖

• 两种依赖关系类型：RDD和它依赖的父RDD的关系有两种不同的类型，即     宽依赖(wide dependency/shuffle dependency)     窄依赖(narrow dependency)

 

• 图解：

 

• 如何区分宽窄依赖：

窄依赖:父RDD的一个分区只会被子RDD的一个分区依赖；
宽依赖:父RDD的一个分区会被子RDD的多个分区依赖(涉及到shuffle)。

2. 为什么要设计宽窄依赖

1. 对于窄依赖：

窄依赖的多个分区可以并行计算；
窄依赖的一个分区的数据如果丢失只需要重新计算对应的分区的数据就可以了。

2. 对于宽依赖：

划分Stage(阶段)的依据:对于宽依赖,必须等到上一阶段计算完成才能计算下一阶段。

六、DAG的生成和划分Stage

1. DAG介绍

• DAG是什么：

DAG(Directed Acyclic Graph有向无环图)指的是数据转换执行的过程，有方向，无闭环(其实就是RDD执行的流程)；
原始的RDD通过一系列的转换操作就形成了DAG有向无环图，任务执行时，可以按照DAG的描述，执行真正的计算(数据被操作的一个过程)。

• DAG的边界

开始:通过SparkContext创建的RDD；
结束:触发Action，一旦触发Action就形成了一个完整的DAG。

2.DAG划分Stage

 

一个Spark程序可以有多个DAG(有几个Action，就有几个DAG，上图最后只有一个Action（图中未表现）,那么就是一个DAG)。

一个DAG可以有多个Stage(根据宽依赖/shuffle进行划分)。

同一个Stage可以有多个Task并行执行(task数=分区数，如上图，Stage1 中有三个分区P1、P2、P3，对应的也有三个 Task)。

可以看到这个DAG中只reduceByKey操作是一个宽依赖，Spark内核会以此为边界将其前后划分成不同的Stage。

同时我们可以注意到，在图中Stage1中，从textFile到flatMap到map都是窄依赖，这几步操作可以形成一个流水线操作，通过flatMap操作生成的partition可以不用等待整个RDD计算结束，而是继续进行map操作，这样大大提高了计算的效率。

• 为什么要划分Stage? --并行计算

一个复杂的业务逻辑如果有shuffle，那么就意味着前面阶段产生结果后，才能执行下一个阶段，即下一个阶段的计算要依赖上一个阶段的数据。那么我们按照shuffle进行划分(也就是按照宽依赖就行划分)，就可以将一个DAG划分成多个Stage/阶段，在同一个Stage中，会有多个算子操作，可以形成一个pipeline流水线，流水线内的多个平行的分区可以并行执行。

• 如何划分DAG的stage？

对于窄依赖，partition的转换处理在stage中完成计算，不划分(将窄依赖尽量放在在同一个stage中，可以实现流水线计算)。

对于宽依赖，由于有shuffle的存在，只能在父RDD处理完成后，才能开始接下来的计算，也就是说需要要划分stage。