Spark之【基础介绍】

Spark最初是由美国伯克利大学AMP实验室在2009年开发，Spark时基于内存计算的大数据并行计算框架，可以用于构建大型的、低延迟的数据分析应用程序。

Spark是当今大数据领域最活跃、最热门、最高效的大数据通用计算平台之一。

Spark的特点

• 运行速度快 ：Spark使用现金的DAG（Directed Acyclic Graph，有向无环图）执行引擎，以支持循环数据流与内存计算，基于内存的执行速度可比Hadoop MapReduce快百倍，基于磁盘的执行速度也能快十倍；
• 容易使用：Spark支持使用Java、Python以及scala等编程语言，简洁的API有助于用户轻松构建并行程序；
• 通用性：Spark提供了完整而强大的技术栈，包括SQL查询、流式计算、机器学习以及GraphX图计算组件，这些组件可以无缝整合在同一个应用中，足以应对复杂的计算；
• 运行模式多样：Spark可以独立运行集群模式中，或者运行在Hadoop中，也可以运行在EC2等云环境中，可以访问HDFS、Hbase、Hive等多种数据源。

Spak相对于Hadoop的优势

Hadoop虽然被广泛运用于大数据建设中，但是本身存在着很大的缺陷，最主要的缺陷是其MapReduce计算模型延迟过高，无法胜任实时、快速计算的需求，因而只适用于离线批处理的场景。

Hadoop的缺点：

• 表达能力有限：计算都必须转换成Map和Reduce两个操作，但这并不适合所有的情况，难以描述复杂的数据处理过程；
• 磁盘IO开销大：每次执行时都需要从磁盘读取数据，并且在计算完成过后要将中间结果写入到磁盘中，IO开销较大；
• 延迟高：一次计算可能需要分解成一系列按照顺序执行的MapReduce任务，任务之间衔接由于干涉到IO开销，会产生较高的延迟。而且，在前一个任务执行完成之前，其他任务无法开始，难以胜任复杂、多阶段的计算任务。

Spark优点：

• 编程模式灵活：Spark计算模式 也属于MapReduce，但不局限于MapReduce操作，还提供了多种数据集操作类型，编程模型也比MapReduce更加灵活；
• Spark内存计算：Spark提供了内存计算，中间结果直接放到内存中，带来了更高的迭代运算效率；
• Spark基于DAG的任务调度执行制度，要优于MapReduce的迭代执行机制。

Spark最大的特点就是将计算数据、中间结果都存储在内存中，大大较少了IO开销；

Spark提供了多种高层次的、简洁的API，通常情况下，对于实现相同功能的应用程序，Spark的代码量要比Hadoop少2-5倍；

但Spark并不能完全替代Hadoop，主要用于替代MapReduce计算模型。实际上，Spark已经很好的融入Hadoop生态圈，并且称为其中重要的一员，它可以借助YARN实现资源调度管理，借助HDFS实现分布式存储。

Spark生态

• Spark Core：Spark Core包含Spark的基本功能，如内存计算、任务调度、部署模式、故障恢复、存储管理等。Spark建立在统一的抽象RDD上，使其可以以基本一致的方式应对不同的大数据处理场景；通常所说的Apache Spark，就是指Spark Core。
• Spark SQL：Spark SQL允许开发人员直接处理RDD，同时也可查询Hive、Hbase等外部数据源。SparkSQL的一个重要特点是其能够统一处理关系表和RDD，使得开发人员可以轻松的使用SQL命令进行查询，并进行更复杂的数据分析；
• Spark Streaming：Spark Streaming支持高吞吐量、可容错处理的实时数据流处理，其核心思路是将流式计算分解成一系列短小的批处理作业。Saprk Streaming支持多种数据输入源，如Kafka、Flume等；
• MLLib（机器学习）：MLLib提供了常用机器学习算法的实现，包括聚类、分类、回归协同过滤等，降低了机器学习的学习门槛，开发人员只要具备一定的理论知识就可以进行机器学习的工作；
• GraphX（图计算）：GraphX是Spark中用于图计算的API，可以认为是Pregel在Spark上重写以及优化，GraphX性能良好，拥有丰富的功能和运算符，能在海量数据上自如的运算复杂的图算法。

Spark的基础概念

• RDD：弹性分布式数据集（Resilient Distributed Dataset）的简称，是分布式内存的一个抽象概念，提供了一种高度受限的共享内存模型；RDD是一个可以容错且并行的数据结构（可以理解成分布式的集合，操作起来和操作本地集合一样简单），它可以将数据集保存在内存中，并且通过控制数据集的分区来达到数据存放处理最优化。代表一个不可变、可分区、里面的元素可并行计算的集合。 1. a list of partitions：一组分片列表，即数据集的基本组成单位。对于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：可选项，对于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节点来进行任务计算。6. 总结：分区列表、分区函数、最佳位置，这三个属性其实说的就是数据集在哪，在哪计算更合适，如何分区；计算函数、依赖关系，这两个属性其实说的是数据集怎么来的。
• DAG：有向无环图，是Directed Acyclic Graph的简称，反映了RDD之间的依赖关系；
• Executor：是运行在工作节点（Worker Node）上的一个进程，负责运行任务，并为应用程序存储数据；
• 应用：用户编写的Spark应用程序；
• 任务：运行在Executor上的工作单元；
• 作业：一个作业包含多个RDD以及作用于响应RDD上的各种操作；
• 阶段：是作业的基本调度单位，一个作业会分为多组任务，每个任务被称为阶段，或者被称为“任务集”。

Spark结构设计

Spark运行架构包括集群资源管理器（Cluster Manager）、运行作业任务的工作节点（Worker Node） 、每个应用的任务控制节点（Driver）和每个工作节点上负责具体任务的执行进程（Executor）。其中，集群资源管理器可以是Spark自带的资源管理器，也可以是Yarn或Mesos等资源管理框架。

Spark各种概念之间的关系

在Spark中，一个应用（Application）由一个任务控制节点和若干个作业（Job）构成，一个作业由多个阶段（Stage）构成，一个阶段由多个任务（Task）组成。当执行一个应用时，任务控制节点会向集群管理器（Cluster Manager）申请资源，启动Executor，并向Executor发送应用程序代码和文件，然后在Executor上执行任务， 运行结束后，执行结果会返回给任务控制节点，或者写到HDFS或者其他数据库中。

Executor的优点

与Hadoop MapReduce计算框架相比，Spark所采用的Executor有两个优点：

1. 利用多线程来执行具体的任务（Hadoop MapReduce采用的是进程模型），减少任务的启动开销；
2. Executor中有一个BlockManager存储模块，会将内存和磁盘共同作为存储设备，当需要多轮迭代计算时，可以将中间结果存储到这个存储模块里，下次需要时，就可以直接读取该存储模块里的数据，而不需要读写到HDFS等文件系统里，因而有效减少了IO开销；或者在交互式查询场景下，预先将表缓存到该存储系统上，从而可以提高读写IO性能。

Spark运行架构的特点：

1. 每个应用都有自己专属的Executor进程，并且该进程在应用运行起家一直驻留。Executor进程以多线程的方式运行任务，减少多进程任务频繁的启动开销，使得任务执行变得非常高效和可靠；
2. Spark运行过程与资源管理器无关，只要能够获取Executor进程并保持通信即可；
3. Executor上有一个BlockManager存储模块，类似于键值存储系统（把内存和磁盘共提供作为存储设备），在处理迭代计算任务时，不需要把中间结果写入到hdfs等文件系统，而是直接放在这个存储系统上，后续有需要时就可以直接读取；在交互查询场景下，也可以把表提前缓存到这个存储系统上，提高读写IO性能；
4. 任务采用了数据本地性和瑞测执行等优化机制。数据本地性是尽量将计算移到数据所在的节点上进行，即“计算向数据靠拢”，因为移动计算比移动数据所占的网络资源要少得多。而且，Spark采用了延时调度机制，可以在更大的程度上实现执行过程优化。比如，拥有数据的节点当前整被其他任务占用，那么，在这种情况下是否需要将数据移动到其他的空闲节点呢？答案是不一定，因为，如果经过预测发现当前节点结束前任务的时间要比移动数据的时间要少，那么，调度就会等待，直到当前节点可用。

RDD持久化/缓存

    某些RDD计算或转换可能会比较耗费时间，如果这些RDD后续还会频繁的被使用到，那么可以将这些RDD进行持久化/缓存；

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

    通过查看RDD的源码发现cache最终也调用了presist无参方法（默认存储只存在内存中）。

小结：

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

RDD容错机制CheckPoint

持久化的局限性：

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

解决方案：

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

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

持久化和Checkpoint的区别：

• 位置：Presist和cache只能保存在本地的磁盘和内存中（或者堆外内存）；Checkpoint可以保存数据到HDFS这类可靠的存储上；
• 生命周期：Cache和Presist的RDD会在程序结束后被清除或者手动调用unpersist方法checkpoint的RDD在程序结束后依然存在，不会被删除。

RDD的依赖关系

    RDD有两种依赖，分别为宽依赖和窄依赖。

• 窄依赖：父RDD的一个分区只会被一个子RDD的分区依赖； 1. 窄依赖的多个分区可以并行计算；2. 窄依赖的一个分区的数据如果丢失只需要重新计算对应的分区的数据就可以了
• 宽依赖：父RDD的一个分区会被子RDD的多个分区依赖（涉及到shuffle） 1. 划分Stage（阶段）的依据：对于宽依赖，必须等待上一个阶段计算完成次啊能计算下一个阶段。

DAG的生成和划分Stage

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

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

DAG划分Stage

• 一个Spark程序可以有多个DAG（有几个Action，就有几个DAG）；
• 一个DAG可以有多个Stage（根据宽依赖/shuffle进行划分）；
• 同一个Stage可以有多个Task并行执行（task数= 分区数）；
• DAG中遇到reduceByKey操作（宽依赖），Spark内核以此为界将其划分成不同的Stage
• flatMap、Map都是窄依赖，这些转换可以形成一个流水操作，通过flatMap操作生成partition可以不用等待整个RDD计算结束，而是继续进行map操作，这样可以大大提高了计算的效率。

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

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

如何划分DAG的stage

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

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

• 总结：Spark会根据shuffle/宽依赖使用回溯算法来对DAG进行Stage划分，从后往前，遇到宽依赖就断开，遇到窄依赖就把当前的RDD加入到当前的stage/阶段中。