1.1 Spark：一体化、多元化的高速大数据通用计算平台和库

基于RDD，Spark成功构建起了一体化、多元化的大数据处理体系，在“One Stack to rule them all”思想的引领下，Spark成功使用Spark SQL、Spark Streaming、MLlib、GraphX近乎完美地解决了大数据中Batch Processing、Streaming Processing、Ad-hoc Query等三大核心问题，更为优雅的是，在Spark中，Spark SQL、Spark Streaming、MLlib、GraphX四大子框架和库之间可以无缝地共享数据和操作，这是当今任何大数据平台都无可匹敌的，Spark的技术堆栈如图1-1所示。

在实际的生产环境中，世界上已经出现很多一千个以上节点的Spark集群，以eBay为例，eBay的Spark集群节点已经超过2000个，Yahoo！等公司也在大规模地使用Spark，国内的淘宝、腾讯、百度、网易、京东、华为、大众点评、优酷土豆等也在生产环境下深度使用Spark。2014 Spark Summit上的信息显示，Spark已经获得世界20家顶级公司的支持，这些公司中包括Intel、IBM等，同时更重要的是，最大的4个Hadoop发行商都为Spark提供了非常强有力的支持。

Apache Spark官方的定义为：Spark是一个通用的大规模数据快速处理引擎。可以简单理解为Spark就是一个大数据分布式处理框架。2009年，Spark作为一个研究项目诞生于伯克利大学AMPLab，最初属于伯克利大学的研究性项目，Spark项目本身使用 Scala 开发（Scala 也是 Spark 平台上的应用程序首选的开发语言），其最初的核心部分代码只有 63 个 Scala 文件，框架实现极为轻量和优雅。Spark于2010年正式开源，于2013年成为Aparch基金项目，并于2014年成为Aparch基金的顶级项目，整个过程不到五年时间。尤其自 2013 年以来，Spark 进入了一个高速发展期，代码库提交与社区活跃度都有显著增长。以活跃度论，Spark在所有Aparch基金会开源项目中位列前三，相较于其他大数据平台或框架而言Spark的代码库最为活跃，如图1-2所示。

Spark在整个发展过程中都烙上了鲜明的学术研究的标记，Spark的基石RDD、Spark Streaming、Spark SQL、MLlib等功能都脱胎于学术研究论文：

● Discretized Streams：Fault-Tolerant Streaming Computation at Scale.Matei Zaharia，Tathagata Das，Haoyuan Li，Timothy Hunter，Scott Shenker，Ion Stoica.SOSP 2013.November 2013.

● Shark：SQL and Rich Analytics at Scale.Reynold Xin，Joshua Rosen，Matei Zaharia，Michael J.Franklin，Scott Shenker，Ion Stoica.SIGMOD 2013.June 2013.

● Discretized Streams：An Efficient and Fault-Tolerant Model for Stream Processing on Large Clusters.Matei Zaharia，Tathagata Das，Haoyuan Li，Scott Shenker，Ion Stoica.HotCloud 2012.June 2012.

● Shark：Fast Data Analysis Using Coarse-grained Distributed Memory（demo）.Cliff Engle，Antonio Lupher，Reynold Xin，Matei Zaharia，Haoyuan Li，Scott Shenker，Ion Stoica.SIGMOD 2012.May 2012.Best Demo Award.

● Resilient Distributed Datasets：A Fault-Tolerant Abstraction for In-Memory Cluster Computing.Matei Zaharia，Mosharaf Chowdhury，Tathagata Das，Ankur Dave，Justin Ma，Murphy McCauley，Michael J.Franklin，Scott Shenker，Ion Stoica.NSDI 2012.April 2012.Best Paper Award and Honorable Mention for Community Award.

● Spark：Cluster Computing with Working Sets.Matei Zaharia，Mosharaf Chowdhury，Michael J.Franklin，Scott Shenker，Ion Stoica.HotCloud 2010.June 2010.

Spark的学术基因使得它从一开始就在大数据领域建立了领先优势，相比于传统的以Hadoop为基石的第一代大数据技术生态系统而言，Spark无论是性能还是方案的统一性都具有极为显著的优势。Spark提供的基于RDD的一体化解决方案，将MapReduce、Streaming、SQL、Machine Learning、Graph Processing等模型统一到一个平台下，并以一致的API公开且提供相同的部署方案，使得Spark的工程应用领域变得更加广泛。

从2013年6月到2014年6月，参与源码贡献的Spark开发人员从原来的68位增长到255位，参与贡献的公司也从17家上升到50家。在这50家公司中，有来自中国的阿里巴巴、百度、网易、腾讯、搜狐等公司。当然，代码库的代码行也从原来的63000行增加到175000行。图1-3为截止到2014年Spark代码贡献者每个月的增长曲线。

图1-4显示了自Spark将其代码部署到Github之后的提交数据，一共有8471次提交，11个分支，25次发布，326位代码贡献者。

目前的Spark最新版本为1.1.0。在该版本的代码贡献者列表中，出现了数十位国内程序员的身影，这些贡献者的多数工作主要集中在Bug Fix上，甚至包括Example的Bug Fix。由于1.1.0版本极大地增强了Spark SQL和MLib的功能，因此有部分贡献都集中在SQL和MLib的特性实现上。

Spark非常重视社区活动，会定期或不定期地举行与Spark相关的会议。会议分为两种，一种是全球Spark顶尖技术人员峰会Spark Summit，已经于2013年和2014年在San Francisco连续召开了两届Summit大会。2015年，Spark Summit将分别在New York与San Francisco召开，其官方网站为：http://spark-summit.org；除了影响力巨大的Spark Summit之外，Spark社区还不定期在全球各地召开小型的Meetup活动。Spark Meetup Group已经遍布北美洲、欧洲、亚洲和大洋洲。在中国，北京Spark Meetup已经召开了两次，并于2014年10月26日召开第三次Meetup。图1-5为Spark Meetup Groups在全球的分布图。

根据Spark的路线图，Databricks会在近三个月陆续发布1.2.0和1.3.0版本。其中，1.2.0版本会对存储方面的API进行重构，尤其值得关注是正在设计和实现当中的外部数据源API，借助这套外部数据源API，用户可以在Spark SQL中采用统一的方式注册和查询来自多种外部数据源的数据，Cassandra、HBase等系统的深度集成将更加统一和高效。在1.3.0之上的版本，则会推出结合Spark和R的SparkR。除了前面提到的SQL与MLlib之外，未来的Spark对于Streaming、GraphX都有不同程度的增强，并能够更好地支持Yarn。

1.1.1 为什么需要使用Spark

1.Hadoop衰落，Spark称霸

由于Hadoop的MapReduce具有高延迟的致命弱点，导致Hadoop无力处理很多对时间有要求的场景，人们对其批评越来越多，Hadoop无力改变现状而导致其正在加速死亡。正如任何领域一样，死亡是一个过程，Hadoop正在示例这样的一个过程，Hadoop的死亡过程在2012年已经开始：

● Hadoop商业版本发行商纷纷宣布支持和转投Spark；

● Mahout前一阶段表示从现在起他们将不再接受任何形式的以MapReduce形式实现的算法，另外一方面，Mahout宣布新的算法基于Spark；

● Cloudera的机器学习框架Oryx的执行引擎也由Hadoop的MapReduce替换成Spark；

● Google已经开始将负载从MapReduce转移到Pregel和Dremel上；

● FaceBook则将负载转移到Presto上；

● 世界头号Hadoop发行和技术支撑厂商Cloudera于2014年7月3号宣布除了Impala支撑部分，一切都将迁移至Spark，具体详情可以参考：https://gigaom.com/2014/07/03/cloudera-impalas-it-for-interactive-sql-on-hadoop-but-everything-else-will-move-to-spark/。

以Yahoo！为例，现阶段的Yahoo！使用的是Hadoop和Spark并存的架构，而随着时间的推移，Spark本身流处理、图技术、机器学习、SQL交互式处理日臻完善，Spark对NoSQL整合和性能更趋完美，最终Yahoo！可能会用Spark全面取代Hadoop，而这也代表了所有做云计算大数据公司的趋势。

或许有朋友会问，Hadoop为何不改进自己？

其实，Hadoop社区一直在改进Hadoop本身，但事实上却是无力回天：

（1）Hadoop的改进基本停留在代码层次，也就是修修补补的事情，这就导致了Hadoop现在具有深度的“技术债务”，负载累累。

（2）Hadoop本身的计算模型决定了Hadoop上所有的工作都要转化成Map、Shuffle和Reduce等核心阶段，由于每次计算都要从磁盘读或者写数据，同时整个计算模型需要网络传输，这就导致了越来越不能忍受的延迟性，同时在前一个任务运行完之前，任何一个任务都不可以运行，这直接导致了其无力支持交互式应用。

那么，为什么不进行重构来实现一个更好的Hadoop呢？答案是Spark的出现使得这样做没有必要性了。

2.为什么需要Spark

2014年10月10号，Spark的商业支持公司DataBricks在其官方博客上对外发布了一条颠覆大数据业界的震撼新闻：在尚未使用Spark内存迭代计算优势的情况下，仅仅使用十分之一计算资源，相同数据的排序上，Spark比Hadoop MapReduce快三倍，Spark彻底颠覆了Hadoop MapReduce保持的排序记录！据Sort Benchmark最新官方消息表明，2014年Sort Benchmark的排序比赛中，Databricks的Spark与加州大学圣地亚哥分校的TritonSort两个系统在2014 Daytona GraySort排序比赛上并列第一，如图1-6所示，具体信息可以访问http://sortbenchmark.org。

这次测试是DataBricks与AWS一起完成的Sort Benchmark（Daytona Gray类别）测试，Sort Benchmark测试是一个考量系统排序100 TB数据（万亿条记录）速度的行业基准测试。在此之前，这项基准测试的世界记录保持者是雅虎，使用2100节点的Hadoop MapReduce集群在72分钟内完成计算，而Spark在使用了206个EC2节点的情况下却将排序用时缩短到了23分钟。此外，在没有官方PB排序对比的情况下，DataBricks首次将Spark推到了1 PB数据（十万亿条记录）的排序，测试的结果是在使用190个节点的情况下，工作负载在短短不到4小时内完成，同样远超雅虎之前使用3800台主机耗时16个小时的记录。一个额外的惊喜是，这也是公用云环境首次完成的PB级排序测试。关于本次测试以及Spark在排序上与Hadoop MapReduce的对比如图1-7所示。

击败Hadoop MapReduce集群创造的大规模数据处理记录证明在任意规模的数据处理中，Spark在性能和扩展性上都更具优势，Spark已经用无懈可击的事实说明使用Spark可以带来时间和开销上的双节省。

从个人学习的角度来看，Spark具有的以下特征也成为你必须掌握Spark的原因。

快速

基于Spark内存的计算速度比Hadoop MapReduce快100倍以上，基于磁盘的计算速度也要快10倍以上。原因是Spark有一个 Directed Acyclic Graph（DAG 有向无环图）执行引擎，支持循环数据流和内存计算，当然这只是性能上的表现。对于学习成本来说，由于Spark提供了一个完整的技术栈，这使得我们能大大节省学习成本，也大大节省运维成本，而时间成本则更值得关注。

易用

提供多语言（包括Scala、Java、Python）API，能够快速实现应用，相比MapReduce 有更简洁的代码，安装部署也无须复杂的配置。使用API可以轻松地构建分布式应用程序，当然也可以使用Scala和Python脚本进行交互式编程。

通用

提供了一个强大的技术栈，包括查询语言Spark SQL、机器学习工具MLlib、图计算工具GraphX、实时流处理工具Spark Streaming，在同一个应用程序中可以无缝地连接使用这些框架，而Hadoop的技术栈则相对独立也较为复杂，各个框架都是独立的系统，给集成带来了很大的复杂性和不确定性，Spark的目的在于构建一个结构上一体化、功能上多元化的高效数据流水线技术栈。

集成Hadoop

Spark可以运行在第二代Hadoop集群管理Yarn上，这使得Spark可以读取Hadoop的任何数据，如果你已经有第二代Hadoop集群，那么你无须安装即可以直接运行Spark，Spark 很容易运行在standalone或者EC2或者Mesos上。它能读取HDFS、HBase、Cassandra以及任何Hadoop数据源。

活跃的社区

Spark起源于2009年，目前已有超过50个机构250个工程师贡献过代码，和2013年6月相比，代码行数几乎扩大3倍，Spark依然处于活跃的开发过程中，代码变动频繁，截至目前已经发布了1.1.0版本。

3.Spark术语

接下来先了解下Spark相关术语，这将有助于我们更好地了解后续的内容。首先我们要说的是Spark的运行模式，Spark有很多种模式，最简单的就是单机本地模式，还有单机为分布式模式，复杂的则运行在集群中，目前能很好地运行在Yarn和Mesos中。当然Spark还有自带的Standalone模式，对于大多数情况Standalone模式就足够了，如果企业已经有Yarn或者Mesos环境，也是很方便部署的，不过本节讨论的是最简单的本地模式。Spark的几种运行模式和常见术语如表1-1和表1-2所示。

每个Spark程序都包含一个驱动程序，而这个驱动程序的任务就是执行用户的main函数，并且将各种算子分布到整个Spark集群中并行运行。

1.1.2 Spark技术生态系统简介

Spark Ecosystem如图1-8所示。

SparkCore包含Spark的基本功能，这些功能包括任务调度、内存管理、故障恢复以及存储系统的交互等，其中包含几个重要的概念：RDD、Stage、DAG。Spark Core的核心思想就是将数据集缓存在内存中，并用Lineage机制来进行容错。

1.Spark Core

1）RDD 功能特性

RDD是弹性分布式数据集（Resilient Distributed Datasets）的简称，是分布式只读的且已分区的集合对象。这些集合是弹性的，如果数据集的一部分丢失，则可以对它们进行重建。具有自动容错、位置感知调度和可伸缩性，而容错性是最难实现的。大多数分布式数据集的容错性有两种方式：数据检查点和记录数据的更新。对于大规模数据分析系统，数据检查点操作成本很高，主要原因是大规模数据在服务器之间的传输会带来各方面的问题，相比记录数据的更新，RDD也只支持粗粒度的转换，也就是记录如何从其他RDD转换而来（即Lineage），以便恢复丢失的分区。

2）RDD编程接口

Spark提供了丰富的API来操作这些数据集，RDD包含2类API：

● Transformations——转换操作，返回值还是一个RDD，如map、filter、union；

● Actions——行动操作，返回结果或是把RDD持久化起来，如count、collect、save。

transformation采用的是懒策略，如果只是将transformation提交是不会提交任务来执行的，任务执行只有在action被提交时才被触发。

Spark支持的RDD转换和动作如表1-3所示。

3）RDD依赖关系

RDD之间的依赖有以下两种：

● 窄依赖（Narrow Dependency）——一个父RDD最多被一个子RDD引用，如map、filter、union操作；

● 宽依赖（Wide Dependency）——一个父 RDD 被多个子 RDD 引用，如 groupByKey操作。

RDD依赖图如图1-9所示。

4）Stage DAG

我们知道Spark提交Job之后会把Job生成多个Stage，多个Stage之间是有依赖的，正如我们前面看到的Stage0依赖State1，Stage之间的依赖关系就构成了DAG（有向无环图）。对于窄依赖，Spark会尽量多地将 RDD 转换放在同一个Stage中；而对于宽依赖，大多数时候是shuffle操作，因此Spark会将此Stage定义为 ShuffleMapStage，以便于向 MapOutputTracker 注册shuffle操作。Spark通常将shuffle操作定义为stage的边界。

Stage划分图如图1-10所示。

2.Spark Streaming

Spark Streaming是一个对实时数据流进行高通量、容错处理的流式处理系统，可以对多种数据源（如Kdfka、Flume、Twitter、Zero和TCP 套接字）进行map、reduce、join、window等复杂操作，并将结果保存到外部文件系统、数据库，或应用到实时仪表盘。

构建在Spark上处理Stream数据的框架，基本的原理是将Stream数据分成小的时间片断（几秒），以类似batch批量处理的方式来处理这一小部分数据。Spark Streaming构建在Spark上，一方面是因为Spark的低延迟执行引擎（100 ms+）可以用于实时计算，另一方面相比基于Record的其他处理框架（如Storm），RDD数据集更容易做高效的容错处理。此外小批量处理的方式使得它可以同时兼容批量和实时数据处理的逻辑和算法，方便了一些需要历史数据和实时数据联合分析的特定应用场合。

Spark Streaming示意图如图1-11所示。

3.Spark Graphx

关于图计算，首先要提到的是Google的Pregel，它是Google继MapReduce之后提出的又一个计算模型，不过Pregel主要用于图的计算。

而Spark早在0.5版本就带了一个还属于实验性的小型Bagel模块，提供了类似Pregel的功能。直到2014年Spark 1.0发布，GraphX才正式投入生产使用。从官网的说明来看，GraphX是Spark中用于图和图并行计算的API，可以认为是GraphLab和Pregel在Spark上的重写及优化。跟其他分布式图计算框架相比，GraphX最大的贡献是在Spark之上提供了一栈式数据解决方案，可以方便且高效地完成图计算的一整套流水作业。

GraphX的核心抽象是Resilient Distributed Property Graph，一种点和边都带属性的有向多重图。它扩展了Spark RDD的抽象，有Table和Graph两种视图，而只需要一份物理存储。两种视图都有自己独有的操作符，从而提高了操作灵活性和执行效率。

Table视图将图看成Vertex Property Table和Edge Property Table等的组合，这些Table继承了Spark RDD的API，如fiter、map等。

Graph视图包括reverse、subgraph、mapV(E)、joinV(E)、mrTriplets等操作。

GraphX也提供了一套图算法工具包，方便用户对图进行分析。GraphX目前依然处于快速发展中。

GraphX的架构如图1-12所示。

4.Spark Shark

Shark即Hive on Spark，本质上是通过Hive的HQL解析，Shark在Spark 1.0.0由于整合度和性能等原因，已经被Spark官方团队停止开发。Shark在实现上是把HQL翻译成Spark上的RDD操作，然后通过Hive的metadata获取数据库里的表信息，Shark获取HDFS上的数据和文件并放到Spark上运算。Shark的最大特性就是快以及与Hive完全兼容。

Shark使用了Hive的API来实现query Parsing和 Logic Plan generation，最后的Physical Plan execution阶段用Spark代替Hadoop MapReduce。通过配置Shark参数，Shark可以自动在内存中缓存特定的RDD，实现数据重用，进而加快特定数据集的检索。同时，Shark通过UDF（用户自定义函数）实现特定的数据分析学习算法，使得SQL数据查询和运算分析能结合在一起，最大化RDD的重复使用。

对比Shark架构图和Hive架构图可以发现Shark为了实现Hive兼容，在HQL方面重用了Hive中HQL的解析、逻辑执行计划翻译、执行计划优化等逻辑，可以近似认为仅将物理执行计划从MR作业替换成了Spark作业（辅以内存列式存储等各种和Hive关系不大的优化）；同时还依赖Hive Metastore和Hive SerDe（用于兼容现有的各种Hive存储格式）。这一策略导致了两个问题，第一是执行计划优化完全依赖于Hive，不方便添加新的优化策略；二是因为MR是进程级并行，写代码的时候不是很注意线程安全问题，导致Shark不得不使用另外一套独立维护的打了补丁的Hive源码分支。

Hive的架构图如图1-13所示。

Shark的架构图如图1-14所示。

5.Spark SQL

Spark SQL是Spark 1.0.0新推出的基于Catalyst引擎的交互式大数据SQL技术，使用SchemaRDD来操作SQL，这个功能和Shark类似，但它比Shark支持更多的查询表达式。最新的版本支持Hive、JSON、Parquet等数据类型，其他数据类型也在计划列表中，如图1-15所示。

6.Spark R

早在2014年一月底，加州大学伯克利分校的AMPLab就发布了其SparkR项目的开发者预览版，该版本使用了基于原生R语言的Apache Spark，但是其使用难度还是在不断增加。SparkR应该被看做R版Spark的轻量级前端，这意味着它不会拥有像Scala或Java那样广泛的API，但它还是能够在R里运行Spark任务和操作数据。Spark通过RDD类提供Spark API，并且允许用户使用R交互式方式在集群中运行任务。它的其中一项关键特性就是有能力序列化闭包，从而能依次透明地将变量副本传入需要参与运算的Spark集群。SparkR还通过内置功能的形式集成了其他R模块，这一功能会在需要某些模块参与运算时通知Spark集群加载特定的模块，但是，不同于闭包，这个需要手动设置。SparkR计划集成MLlib机器学习类库，这将有益于MLlib的发展。

7.MLBase

机器学习和统计学技术是把大数据转为行为知识的关键点。对于大多数开发者而言，要掌握机器学习的相关知识不是一件容易的事情，即使能理解算法本身，而实现又是一个难题，更重要的是，企业随着数据的积累，并行化算法实现的需求越来越多，对应多种多样的数据，算法参数的设置和调优需要很强的领域经验，从而导致机器学习成为一个神秘的学科。MLBase的愿景就在于扫除这些障碍，开发者只要具备一定的算法理论知识，即可进行机器学习的工作。

MLBase提供了一个简单的声明式方法来指定机器学习任务，并能动态地选择最优的学习算法，这大大简化了开发者的工作。

MLBase包含了高级别抽象和优化器以及MLlib算法库，MLlib 是增长最快的机器学习类库之一，它拥有超过137个的贡献者。说明如下：

ML Optimizer——优化器会选择最适合的、已经实现好的机器学习算法和相关参数；

MLI——一个进行特征抽取和高级ML编程抽象的算法实现的API或平台；

MLlib——基于Spark的底层分布式机器学习库，可以不断地扩充算法；

MLRuntime——基于Spark计算框架，将Spark的分布式计算应用到机器学习领域。

MLBase的架构如图1-16所示。

Spark直接可用的机器学习子框架MLlib是MLbase的一部分，MLbase的架构如图1-17所示。

8.Tachyon

Tachyon是一个高容错的分布式文件系统，允许文件以内存的速度在集群框架中进行可靠的共享，就像Spark和 MapReduce那样。通过信息继承、内存侵入，Tachyon获得了高性能。Tachyon工作集文件缓存在内存中，并且让不同的 Jobs/Queries以及框架都能内存的速度来访问缓存文件。因此，Tachyon可以减少那些需要经常使用的数据集通过访问磁盘来获得的次数。Tachyon兼容Hadoop，现有的Spark和MR程序不需要任何修改即可运行，Tachyon与大数据构建的技术堆栈如图1-18所示。

2013年4月，AMPLab共享了其Tachyon 0.2.0 Alpha版本的Tachyon，宣称其性能为HDFS的300倍，继而受到了极大的关注。Tachyon的几个特性如下：

JAVA-Like File API——Tachyon提供类似Java File类的API；

兼容性——Tachyon实现了HDFS接口，所以Spark和MR程序不需要任何修改即可运行；

可插拔的底层文件系统——Tachyon是一个可插拔的底层文件系统，提供容错功能。Tachyon将内存数据记录在底层文件系统中。它有一个通用的接口，使得其可以很容易地插入到不同的底层文件系统中。目前支持 HDFS、S3、GlusterFS 和单节点的本地文件系统，未来将支持更多的文件系统。

9.BlinkDB

BlinkDB 是一个用于在海量数据上运行交互式 SQL 查询的大规模并行查询引擎。它允许用户通过权衡数据精度来提升查询响应时间，其数据的精度被控制在允许的误差范围内。为了达到这个目标，BlinkDB 有两个核心思想：

自适应优化框架——原始数据随着时间的推移建立并维护一组多维样本；

动态样本选择策略——基于查询的准确性和（或）响应时间需求来选择一个适当大小的样本。

BlinkDB的架构如图1-19所示。

10.Akka

Akka是一个平台，灵感来自ERlang却使用Scala语言实现，能更轻松地开发可扩展，实现多线程安全应用。虽然在大多数流行的语言中并发是基于多线程之间的共享内存，使用同步方法防止写争夺，但Akka提供的并发模型基于Actors。Actors是一个轻量级的对象，通过发送消息实现交互。每个Actors在同一时间处理最多一个消息，可以发送消息给其他Actors。Akka具体的架构示意图如图1-20、图1-21、图1-22所示。