Spark详解

?

Spark概述

当前，MapReduce编程模型已经成为主流的分布式编程模型，它极大地方便了编程人员在不会分布式并行编程的情况下，将自己的程序运行在分布式系统上。但是MapReduce也存在一些缺陷，如高延迟、不支持DAG模型、Map与Reduce的中间数据落地等。因此在近两年，社区出现了优化改进MapReduce的项目，如交互查询引擎Impala、支持DAG的TEZ、支持内存计算Spark等。Spark是UC Berkeley AMP lab开源的通用并行计算框架，以其先进的设计理念，已经成为社区的热门项目。Spark相对与MapReduce的优势有：低延迟、支持DAG和分布式内存计算。虽然Spark有许多优势，但是毕竟没有经过大规模生产的验证，所以暂未能代替MapReduce，庆幸的是，由于许多人意识到Spark的优势所在，社区Spark已成热门项目。本文对Spark的分析基于社区spark 1.0.2版本。

Spark生态圈介绍

????Spark力图整合机器学习（MLib）、图算法（GraphX）、流式计算（Spark Streaming）和数据仓库（Spark SQL）等领域，通过计算引擎Spark，弹性分布式数据集（RDD），架构出一个新的大数据应用平台。

????Spark生态圈以HDFS、S3、Techyon为底层存储引擎，以Yarn、Mesos和Standlone作为资源调度引擎；使用Spark，可以实现MapReduce应用；基于Spark，Spark SQL可以实现即席查询，Spark Streaming可以处理实时应用，MLib可以实现机器学习算法，GraphX可以实现图计算，SparkR可以实现复杂数学计算。

?

?

图1 Spark生态圈

Spark包与目录介绍

下载源码包：http://spark.apache.org/downloads.html

Spark 1.0.2源码包目录结构：

图2 spark代码目录结构

源代码下子目录很多，下表是几个关键目录的介绍。

非关键目录

Spark编译

Spark使用mvn编译，并为我们提交了构建项目的脚本：make-distribution.sh,推荐在Linux下编译，编译命令：./make-distribution.sh --hadoop 2.2.0 --with-yarn –tgz

编译成功后会工程目录下生成dist目录，即项目可执行包：

图3

如何运行Spark作业

参考社区文档： https://spark.apache.org/docs/latest/quick-start.html

?

Spark运行模式介绍

Spark任务的运行模式有local、standalone、OnYarn等，各种运行模式的详细流程可以参考博客： http://www.cnblogs.com/shenh062326/p/3658543.html

Spark作业执行简要流程

????无论运行在哪种模式下，Spark作业的执行流程都是相似的，主要有如下八步：

1. 客户端提交作业
2. Driver启动流程
3. Driver申请资源并启动其余Executor(即Container)
4. Executor启动流程
5. 作业调度，生成stages与tasks。
6. Task调度到Executor上，Executor启动线程执行Task逻辑
7. Driver管理Task状态
8. Task完成，Stage完成，作业完成

?

Spark原理详细介绍

DAGScheduler与TaskScheduler详解

DAGScheduler把一个spark作业转换成成stage的DAG（Directed Acyclic Graph有向无环图），根据RDD和stage之间的关系，找出开销最小的调度方法，然后把stage以TaskSet的形式提交给TaskScheduler。图3与图4展示了DAGScheduler与TaskScheduler的工作。

图4 DAGScheduler的作用

图5 TaskScheduler作用

????DAGScheduler还决定了运行task的理想位置，并把这些信息传递给下层的TaskScheduler。此外，DAGScheduler还处理由于shuffle数据丢失导致的失败，这有可能需要重新提交运行之前的stage（非shuffle数据丢失导致的task失败由TaskScheduler处理）。

????TaskScheduler维护所有TaskSet，当Executor向Driver发送心跳时，TaskScheduler会根据其资源剩余情况分配相应的Task。另外TaskScheduler还维护着所有Task的运行状态，重试失败的Task。

RDD详解

RDD（Resilient Distributed Datasets弹性分布式数据集），是spark中最重要的概念，用户可以简单的把RDD理解成一个提供了许多操作接口的数据集合，和一般数据集不同的是，其实际数据分布存储于一批机器中（内存或磁盘中）。当然，RDD肯定不会这么简单，它的功能还包括容错、集合内的数据可以并行处理等。图5是RDD类的视图，图6简要展示了RDD的底层实现。更多RDD的操作描述和编程方法请参考社区文档：https://spark.apache.org/docs/latest/programming-guide.html。

图6 RDD提供了许多操作

?

图7 RDD的实现

RDD cache的原理

RDD的转换过程中，并不是每个RDD都会存储，如果某个RDD会被重复使用，或者计算其代价很高，那么可以通过显示调用RDD提供的cache()方法，把该RDD存储下来。那RDD的cache是如何实现的呢？

RDD中提供的cache()方法只是简单的把该RDD放到cache列表中。当RDD的iterator被调用时，通过CacheManager把RDD计算出来，并存储到BlockManager中，下次获取便可直接通过CacheManager从BlockManager取出。

Shuffle原理简介

在Spark编程时，不仅仅只有reduce才会产生shuffle过程，RDD提供的groupByKey，countApproxDistinctByKey等操作都会生成shuffle。Spark中shuffle的实现与MapReduce的shuffle有比较大的差别，首先是map阶段，map的输出不再需要排序，直接写到文件中，一个map会把属于不同reduce的数据分别输出到不同的文体中，而reduce则通过aggregator处理所有shuffle fetch获取的partition。

从流程上看，MapTask结束后，Driver的MapOutPutTracker会注册MapOutPuts，ReduceTask启动后向Driver获取MapOutPutStatuses，然后fetch相应的MapOutPuts。

图8 Shuffle简介

图9 Shuffle的原理（摘自网络）

Spark Streaming详解

Spark Streaming是建立在Spark上的实时计算框架，通过它提供丰富的API、基于内存的高速执行引擎，用户可以结合流式、批处理和交互试查询应用。

Spark Streaming的基本原理是将输入数据流以时间片（秒级）为单位进行拆分，然后以类似批处理的方式处理每个时间片数据，其基本原理如下图所示。

图10 Spark Streaming基本原理图

首先，Spark Streaming把实时输入数据流以时间片Δt （如1秒）为单位切分成块。Spark Streaming会把每块数据作为一个RDD，并使用RDD操作处理每一小块数据。每个块都会生成一个Spark Job处理，最终结果也返回多块。

使用Spark Streaming编写的程序与编写Spark程序非常相似，在Spark程序中，主要通过操作RDD（Resilient Distributed Datasets弹性分布式数据集）提供的接口，如map、reduce、filter等，实现数据的批处理。而在Spark Streaming中，则通过操作DStream（表示数据流的RDD序列）提供的接口，这些接口和RDD提供的接口类似。

图11 Spark Streaming程序转换为DStream Graph

图12 DStream Graph转换为Spark jobs

在图12中，Spark Streaming把程序中对DStream的操作转换为DStream Graph，图4中，对于每个时间片，DStream Graph都会产生一个RDD Graph；针对每个输出操作（如print、foreach等），Spark Streaming都会创建一个Spark action；对于每个Spark action，Spark Streaming都会产生一个相应的Spark job，并交给JobManager。JobManager中维护着一个Jobs队列, Spark job存储在这个队列中，JobManager把Spark job提交给Spark Scheduler，Spark Scheduler负责调度Task到相应的Spark Executor上执行。

图13

Spark Streaming的另一大优势在于其容错性，RDD会记住创建自己的操作，每一批输入数据都会在内存中备份，如果由于某个结点故障导致该结点上的数据丢失，这时可以通过备份的数据在其它结点上重算得到最终的结果。