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更多幹貨內容請關注微信公衆號「錄信數軟」~