Hadoop是否会被Spark取代？Hadoop生态组件原理解析

Hadoop和Spark都是目前主流的大数据框架，可是随着Spark在速度和易用性方面表现出的优点，一些国内外专家逐渐推崇Spark技术，而且认为Spark才是大数据的将来。本文将会浅析Hadoop生态的发展历程及其中部分组件的技术原理，最终就Hadoop是否会被Spark取代给出结论。

1、Hadoop的核心组件

在对Hadoop核心组件进行介绍以前，咱们须要先了解Hadoop解决了什么问题。Hadoop主要就是解决了大数据（大到一台计算机没法进行存储，一台计算机没法在要求的时间内进行处理）的可靠存储和处理。

Hadoop的核心组件主要有三个，分别是：HDFS、YARN和MapReduce。HDFS是是google三大论文之一的GFS的开源实现，是一个高度容错性的系统，适合部署在廉价的机器上的，适合存储海量数据的分布式文件系统。而YARN是Hadoop的资源管理器，能够视为一个分布式的操做系统平台。相比于HDFS和YARN，MapReduce能够说是Hadoop的核心组件，如下会就MapReduce进行重点讨论。

MapReduce，经过简单的Mapper和Reducer的抽象提供一个编程模型，能够在一个由几十台上百台的PC组成的不可靠集群上并发地，分布式地处理大量的数据集，而把并发、分布式（如机器间通讯）和故障恢复等计算细节隐藏起来。而Mapper和Reducer的抽象，又是各类各样的复杂数据处理均可以分解为的基本元素。这样，复杂的数据处理能够分解为由多个Job（包含一个Mapper和一个Reducer）组成的有向无环图（DAG）,而后每一个Mapper和Reducer放到Hadoop集群上执行，就能够得出结果。

​
在MapReduce中，Shuffle是一个很是重要的过程，正是有了看不见的Shuffle过程，才可使在MapReduce之上写数据处理的开发者彻底感知不到分布式和并发的存在。

广义的Shuffle是指图中在Map和Reuce之间的一系列过程。

2、Hadoop的局限和改进

尽管Hadoop提供了处理海量数据的能力，可是Hadoop的核心组件——MapReduce的使用问题仍是一直困扰着Hadoop的发展，MapReduce的局限主要能够总结为如下几点：

• 抽象层次低，须要手工编写代码来完成，使用上难以上手
• 只提供两个操做，Map和Reduce，表达力欠缺
• 一个Job只有Map和Reduce两个阶段（Phase），复杂的计算须要大量的Job完成，Job之间的依赖关系是由开发者本身管理的
• 处理逻辑隐藏在代码细节中，没有总体逻辑
• 中间结果也放在HDFS文件系统中
• ReduceTask须要等待全部MapTask都完成后才能够开始
• 时延高，只适用Batch数据处理，对于交互式数据处理，实时数据处理的支持不够
• 对于迭代式数据处理性能比较差

好比说，用MapReduce实现两个表的Join都是一个颇有技巧性的过程，以下图所示：

所以，在Hadoop推出以后，出现了不少相关的技术对其中的局限进行改进，如Pig，Cascading，JAQL，OOzie，Tez，Spark等，下面就对一些重要技术进行重点的讨论。

1.Apache Pig

Apache Pig是Hadoop框架中的一部分，Pig提供类SQL语言（Pig Latin）经过MapReduce来处理大规模半结构化数据。而Pig Latin是更高级的过程语言，经过将MapReduce中的设计模式抽象为操做，如Filter，GroupBy，Join，OrderBy，由这些操做组成有向无环图（DAG）。例如以下程序就描述了数据处理的整个过程。

visits       = load ‘/data/visits’ as (user, url, time);
gVisits      = group visits by url;
visitCounts  = foreach gVisits generate url, count(visits);
​
urlInfo      = load ‘/data/urlInfo’ as (url, category, pRank);
visitCounts  = join visitCounts by url, urlInfo by url;
​
gCategories  = group visitCounts by category;
topUrls = foreach gCategories generate top(visitCounts,10);
​
store topUrls into ‘/data/topUrls’;

而Pig Latin又是经过编译为MapReduce，在Hadoop集群上执行的。上述程序被编译成MapReduce时，会产生以下图所示的Map和Reduce：

Apache Pig解决了MapReduce存在的大量手写代码，语义隐藏，提供操做种类少的问题。相似的项目还有Cascading，JAQL等。

2.Apache Tez

Apache Tez，Tez是HortonWorks的Stinger Initiative的的一部分。做为执行引擎，Tez也提供了有向无环图（DAG），DAG由顶点（Vertex）和边（Edge）组成，Edge是对数据的移动的抽象，提供了One-To-One，BroadCast，和Scatter-Gather三种类型，只有Scatter-Gather才须要进行Shuffle。

以以下SQL为例：

SELECT a.state, COUNT(*),
AVERAGE(c.price)
FROM a
JOIN b ON (a.id = b.id)
JOIN c ON (a.itemId = c.itemId)
GROUP BY a.state

图中蓝色方块表示Map，绿色方块表示Reduce，云状表示写屏障（write barrier，一种内核机制，能够理解为持久的写），Tez的优化主要体如今：

去除了连续两个做业之间的写屏障

去除了每一个工做流中多余的Map阶段（Stage）

经过提供DAG语义和操做，提供了总体的逻辑，经过减小没必要要的操做，Tez提高了数据处理的执行性能。

3.Apache Spark

Apache Spark是一个新兴的大数据处理的引擎，主要特色是提供了一个集群的分布式内存抽象，以支持须要工做集的应用。

这个抽象就是RDD（Resilient Distributed Dataset），RDD就是一个不可变的带分区的记录集合，RDD也是Spark中的编程模型。Spark提供了RDD上的两类操做，转换和动做。转换是用来定义一个新的RDD，包括map, flatMap, filter, union, sample, join, groupByKey, cogroup, ReduceByKey, cros, sortByKey, mapValues等，动做是返回一个结果，包括collect, reduce, count, save, lookupKey。

Spark的API很是简单易用，Spark的WordCount的示例以下所示：

val spark = new SparkContext(master, appName, [sparkHome], [jars])
val file = spark.textFile("hdfs://...")
val counts = file.flatMap(line => line.split(" "))
                 .map(word => (word, 1))
                 .reduceByKey(_ + _)
counts.saveAsTextFile("hdfs://...")

其中的file是根据HDFS上的文件建立的RDD，后面的flatMap，map，reduceByKe都建立出一个新的RDD，一个简短的程序就可以执行不少个转换和动做。

在Spark中，全部RDD的转换都是是惰性求值的。RDD的转换操做会生成新的RDD，新的RDD的数据依赖于原来的RDD的数据，每一个RDD又包含多个分区。那么一段程序实际上就构造了一个由相互依赖的多个RDD组成的有向无环图（DAG）。并经过在RDD上执行动做将这个有向无环图做为一个Job提交给Spark执行。

例如，上面的WordCount程序就会生成以下的DAG：

scala> counts.toDebugString
res0: String =
MapPartitionsRDD[7] at reduceByKey at <console>:14 (1 partitions)
  ShuffledRDD[6] at reduceByKey at <console>:14 (1 partitions)
    MapPartitionsRDD[5] at reduceByKey at <console>:14 (1 partitions)
      MappedRDD[4] at map at <console>:14 (1 partitions)
        FlatMappedRDD[3] at flatMap at <console>:14 (1 partitions)
          MappedRDD[1] at textFile at <console>:12 (1 partitions)
            HadoopRDD[0] at textFile at <console>:12 (1 partitions)

Spark对于有向无环图Job进行调度，肯定阶段（Stage），分区（Partition），流水线（Pipeline），任务（Task）和缓存（Cache），进行优化，并在Spark集群上运行Job。RDD之间的依赖分为宽依赖（依赖多个分区）和窄依赖（只依赖一个分区），在肯定阶段时，须要根据宽依赖划分阶段。根据分区划分任务。

Spark支持故障恢复的方式也不一样，提供两种方式，Linage，经过数据的血缘关系，再执行一遍前面的处理，Checkpoint，将数据集存储到持久存储中。

Spark为迭代式数据处理提供更好的支持。每次迭代的数据能够保存在内存中，而不是写入文件。

Spark的性能相比Hadoop有很大提高，2014年10月，Spark完成了一个Daytona Gray类别的Sort Benchmark测试，排序彻底是在磁盘上进行的，与Hadoop以前的测试的对比结果如表格所示：

从表格中能够看出排序100TB的数据（1万亿条数据），Spark只用了Hadoop所用1/10的计算资源，耗时只有Hadoop的1/3。

Spark的优点不只体如今性能提高上的，Spark框架为批处理（Spark Core），交互式（Spark SQL），流式（Spark Streaming），机器学习（MLlib），图计算（GraphX）提供一个统一的数据处理平台，这相对于使用Hadoop有很大优点。

特别是在有些状况下，你须要进行一些ETL工做，而后训练一个机器学习的模型，最后进行一些查询，若是是使用Spark，你能够在一段程序中将这三部分的逻辑完成造成一个大的有向无环图（DAG），并且Spark会对大的有向无环图进行总体优化。

例以下面的程序：

val points = sqlContext.sql(   “SELECT latitude, longitude FROM historic_tweets”)  
​
val model = KMeans.train(points, 10)  
​
sc.twitterStream(...)   .map(t => (model.closestCenter(t.location), 1))   .reduceByWindow(“5s”, _ + _)

这段程序的第一行是用Spark SQL 查寻出了一些点，第二行是用MLlib中的K-means算法使用这些点训练了一个模型，第三行是用Spark Streaming处理流中的消息，使用了训练好的模型。

3、总结

咱们能够借助于逻辑电路来理解MapReduce和Spark。若是说MapReduce是公认的分布式数据处理的低层次抽象，相似逻辑门电路中的与门，或门和非门，那么Spark的RDD就是分布式大数据处理的高层次抽象，相似逻辑电路中的编码器或译码器等。

RDD就是一个分布式的数据集合（Collection），对这个集合的任何操做均可以像函数式编程中操做内存中的集合同样直观、简便，但集合操做的实现倒是在后台分解成一系列Task发送到几十台上百台服务器组成的集群上完成的。最近新推出的大数据处理框架Apache Flink也使用数据集（Data Set）和其上的操做做为编程模型的。

由RDD组成的有向无环图（DAG）的执行是调度程序将其生成物理计划并进行优化，而后在Spark集群上执行的。Spark还提供了一个相似于MapReduce的执行引擎，该引擎更多地使用内存，而不是磁盘，获得了更好的执行性能。

基于此，Spark针对于Hadoop的一些局限进行了解决：

• 抽象层次低，须要手工编写代码来完成，使用上难以上手
  =>基于RDD的抽象，实数据处理逻辑的代码很是简短
• 只提供两个操做，Map和Reduce，表达力欠缺
  =>提供不少转换和动做，不少基本操做如Join，GroupBy已经在RDD转换和动做中实现
• 一个Job只有Map和Reduce两个阶段（Phase），复杂的计算须要大量的Job完成，Job之间的依赖关系是由开发者本身管理的
  =>一个Job能够包含RDD的多个转换操做，在调度时能够生成多个阶段（Stage），并且若是多个map操做的RDD的分区不变，是能够放在同一个Task中进行
• 处理逻辑隐藏在代码细节中，没有总体逻辑
  =>在Scala中，经过匿名函数和高阶函数，RDD的转换支持流式API，能够提供处理逻辑的总体视图。代码不包含具体操做的实现细节，逻辑更清晰
• 中间结果也放在HDFS文件系统中
  =>中间结果放在内存中，内存放不下了会写入本地磁盘，而不是HDFS
• ReduceTask须要等待全部MapTask都完成后才能够开始
  => 分区相同的转换构成流水线放在一个Task中运行，分区不一样的转换须要Shuffle，被划分到不一样的Stage中，须要等待前面的Stage完成后才能够开始
• 时延高，只适用Batch数据处理，对于交互式数据处理，实时数据处理的支持不够
  =>经过将流拆成小的batch提供Discretized Stream处理流数据
• 对于迭代式数据处理性能比较差
  =>经过在内存中缓存数据，提升迭代式计算的性能

所以，Hadoop MapReduce会被新一代的大数据处理平台替代是技术发展的趋势，而在新一代的大数据处理平台中，Spark目前获得了最普遍的承认和支持。

最后，咱们经过Lambda Architecture的一个案例做为结束和补充，Lambda Architecture是一个大数据处理平台的参考模型，以下图所示：

其中包含3层，Batch Layer，Speed Layer和Serving Layer，因为Batch Layer和Speed Layer的数据处理逻辑是一致的，若是用Hadoop做为Batch Layer，而用Storm做为Speed Layer，你须要维护两份使用不一样技术的代码。

而Spark能够做为Lambda Architecture一体化的解决方案,大体以下：

Batch Layer，HDFS+Spark Core，将实时的增量数据追加到HDFS中，使用Spark Core批量处理全量数据，生成全量数据的视图

Speed Layer，Spark Streaming来处理实时的增量数据，以较低的时延生成实时数据的视图

Serving Layer，HDFS+Spark SQL（也许还有BlinkDB），存储Batch Layer和Speed Layer输出的视图，提供低时延的即席查询功能，将批量数据的视图与实时数据的视图合并

经过这个案例，咱们也能够再一次重申上述结论，Spark是能够取代MapReduce从而成为Hadoop系统中不可或缺的一部分，可是没法取代Hadoop生态。

END更多干货内容请关注微信公众号“录信数软”~