Spark 與 Hadoop 學習筆記介紹及對比

2019 年 10 月 6 日
筆記

這篇部落格將會簡單記錄Hadoop與Spark對比，HDFS，MapReduce的基本概念，及Spark架構設計，RDD，運行模式。整理起來一起閱讀方便我們理解整個大數據處理框架全局和發展。

1. Hadoop

1.1 背景

Hadoop就是解決了大數據（大到一台電腦無法進行存儲，一台電腦無法在要求的時間內進行處理）的可靠存儲和處理。

HDFS: 分散式文件存儲

YARN: 分散式資源管理

MapReduce: 分散式計算

Others: 利用YARN的資源管理功能實現其他的數據處理方式

內部各個節點基本都是採用Master-Woker架構

適合

大規模數據
流式數據（寫一次，讀多次）
商用硬體（一般硬體）

不適合

低延時的數據訪問
大量的小文件
頻繁修改文件（基本就是寫1次）1.2 HDFSHadoop Distributed File System，分散式文件系統
Block數據

基本存儲單位，一般大小為64M（配置大的塊主要是因為：1）減少搜尋時間，一般硬碟傳輸速率比尋道時間要快，大的塊可以減少尋道時間；2）減少管理塊的數據開銷，每個塊都需要在NameNode上有對應的記錄；3）對數據塊進行讀寫，減少建立網路的連接成本）
一個大文件會被拆分成一個個的塊，然後存儲於不同的機器。如果一個文件少於Block大小，那麼實際佔用的空間為其文件的大小
基本的讀寫單位，類似於磁碟的頁，每次都是讀寫一個塊
每個塊都會被複制到多台機器，默認複製3份

NameNode

存儲文件的metadata，運行時所有數據都保存到記憶體，整個HDFS可存儲的文件數受限於NameNode的記憶體大小
一個Block在NameNode中對應一條記錄（一般一個block佔用150位元組），如果是大量的小文件，會消耗大量記憶體。同時map task的數量是由splits來決定的，所以用MapReduce處理大量的小文件時，就會產生過多的map task，執行緒管理開銷將會增加作業時間。處理大量小文件的速度遠遠小於處理同等大小的大文件的速度。因此Hadoop建議存儲大文件
數據會定時保存到本地磁碟，但不保存block的位置資訊，而是由DataNode註冊時上報和運行時維護（NameNode中與DataNode相關的資訊並不保存到NameNode的文件系統中，而是NameNode每次重啟後，動態重建）
NameNode失效則整個HDFS都失效了，所以要保證NameNode的可用性

Secondary NameNode

定時與NameNode進行同步（定期合併文件系統鏡像和編輯日期，然後把合併後的傳給NameNode，替換其鏡像，並清空編輯日誌，類似於CheckPoint機制），但NameNode失效後仍需要手工將其設置成主機

DataNode

保存具體的block數據
負責數據的讀寫操作和複製操作
DataNode啟動時會向NameNode報告當前存儲的數據塊資訊，後續也會定時報告修改資訊
DataNode之間會進行通訊，複製數據塊，保證數據的冗餘性
DataNode會定時發送心跳到NameNode。如果一段時間內NameNode沒有收到DataNode的心跳消息，則認為其失效。此時NameNode就會將該節點的數據（從該節點的複製節點中獲取）複製到另外的DataNode中1.3 MapReduceMapReduce是一種編程模型，用於大規模數據集（大於1TB）的並行運算,用於解決海量數據的計算問題。

MapReduce分成了兩個部分：

映射（Mapping）對集合里的每個目標應用同一個操作。即，如果你想把表單里每個單元格乘以二，那麼把這個函數單獨地應用在每個單元格上的操作就屬於mapping。
化簡（Reducing）遍歷集合中的元素來返回一個綜合的結果。即，輸出表單里一列數字的和這個任務屬於reducing。

你向MapReduce框架提交一個計算作業時，它會首先把計算作業拆分成若干個Map任務，然後分配到不同的節點上去執行，

每一個Map任務處理輸入數據中的一部分，當Map任務完成後，它會生成一些中間文件，這些中間文件將會作為Reduce任務的輸入數據。

Reduce任務的主要目標就是把前面若干個Map的輸出匯總到一起並輸出。

MapReduce的偉大之處就在於編程人員在不會分散式並行編程的情況下，將自己的程式運行在分散式系統上。

一切都是從最上方的user program開始的，user program鏈接了MapReduce庫，實現了最基本的Map函數和Reduce函數。圖中執行的順序都用數字標記了。

MapReduce庫先把user program的輸入文件劃分為M份（M為用戶定義），每一份通常有16MB到64MB，如圖左方所示分成了split0~4；然後使用fork將用戶進程拷貝到集群內其它機器上。
user program的副本中有一個稱為master，其餘稱為worker，master是負責調度的，為空閑worker分配作業（Map作業3或者Reduce作業），worker的數量也是可以由用戶指定的。
被分配了Map作業的worker，開始讀取對應分片的輸入數據，Map作業數量是由M決定的，和split一一對應；Map作業從輸入數據中抽取出鍵值對，每一個鍵值對都作為參數傳遞給map函數，map函數產生的中間鍵值對被快取在記憶體中。
快取的中間鍵值對會被定期寫入本地磁碟，而且被分為R個區，R的大小是由用戶定義的，將來每個區會對應一個Reduce作業；這些中間鍵值對的位置會被通報給master，master負責將資訊轉發給Reduce worker。
master通知分配了Reduce作業的worker它負責的分區在什麼位置（肯定不止一個地方，每個Map作業產生的中間鍵值對都可能映射到所有R個不同分區），當Reduce worker把所有它負責的中間鍵值對都讀過來後，先對它們進行排序，使得相同鍵的鍵值對聚集在一起。因為不同的鍵可能會映射到同一個分區也就是同一個Reduce作業（誰讓分區少呢），所以排序是必須的。
reduce worker遍歷排序後的中間鍵值對，對於每個唯一的鍵，都將鍵與關聯的值傳遞給reduce函數，reduce函數產生的輸出會添加到這個分區的輸出文件中。
當所有的Map和Reduce作業都完成了，master喚醒正版的user program，MapReduce函數調用返回user program的程式碼
所有執行完畢後，MapReduce輸出放在了R個分區的輸出文件中（分別對應一個Reduce作業）。用戶通常並不需要合併這R個文件，而是將其作為輸入交給另一個MapReduce程式處理。整個過程中，輸入數據是來自底層分散式文件系統（GFS）的，中間數據是放在本地文件系統的，最終輸出數據是寫入底層分散式文件系統（GFS）的。而且我們要注意Map/Reduce作業和map/reduce函數的區別：Map作業處理一個輸入數據的分片，可能需要調用多次map函數來處理每個輸入鍵值對；Reduce作業處理一個分區的中間鍵值對，期間要對每個不同的鍵調用一次reduce函數，Reduce作業最終也對應一個輸出文件。

2. Spark

2.1 簡介

Apache Spark是一種快速的集群計算技術，是UC Berkeley AMP lab所開源的類Hadoop MapReduce的通用的並行計算框架，專為快速計算而設計。它基於Hadoop MapReduce，擴展了MapReduce模型，以有效地將其用於更多類型的計算，包括互動式查詢和流處理。 Spark的主要特性是它的記憶體中集群計算，從而不再需要讀寫HDFS，提高了應用程式的處理速度，因此Spark能更好地適用於數據挖掘與機器學習等需要迭代的map reduce的演算法。

Spark旨在涵蓋各種工作負載，如批處理應用程式，迭代演算法，互動式查詢和流式處理。除了在相應系統中支援所有這些工作負載之外，它還減少了維護單獨工具的管理負擔。

對於一個普遍的信念，Spark不是Hadoop的修改版本，並不是真的依賴於Hadoop，因為它有自己的集群管理。 Hadoop只是實現Spark的方法之一。

Spark以兩種方式使用Hadoop - 一個是存儲，另一個是處理。由於Spark具有自己的集群管理計算，因此它僅使用Hadoop進行存儲。

基本概念

Application：用戶自己寫的 Spark 應用程式，批處理作業的集合。Application 的 main 方法為應用程式的入口，用戶通過 Spark 的 API，定義了 RDD 和對 RDD 的操作
SparkContext： Spark 最重要的 API，用戶邏輯與 Spark 集群主要的交互介面，它會和 Cluster Master 交互，包括向它申請計算資源等（在Spark2.0新標準中使用SparkSession，其實質為SQLContext與HiveContext，內部封裝SparkContext）
Driver 和 Executor：Spark 在執行每個 Application 的過程中會啟動 Driver 和 Executor 兩種 JVM 進程。Driver 進程為主控進程，負責執行用戶 Application 中的 main 方法，提交 Job，並將 Job 轉化為 Task，在各個 Executor 進程間協調 Task 的調度。運行在Worker上的 Executor 進程負責執行 Task，並將結果返回給 Driver，同時為需要快取的 RDD 提供存儲功能 ###2.2 彈性分散式數據集(RDD) 彈性分散式數據集(RDD)是 Spark 框架中的核心概念。可以將 RDD 視作資料庫中的一張表。其中可以保存任何類型的數據。Spark 將數據存儲在不同分區上的 RDD 之中。

RDD 可以幫助重新安排計算並優化數據處理過程。

此外，它還具有容錯性，因為RDD知道如何重新創建和重新計算數據集。

RDD 是不可變的。你可以用變換（Transformation）修改 RDD，但是這個變換所返回的是一個全新的RDD，而原有的 RDD 仍然保持不變。

RDD 支援兩種類型的操作：

變換（Transformation）變換的返回值是一個新的 RDD 集合，而不是單個值。調用一個變換方法，不會有任何求值計算，它只獲取一個 RDD 作為參數，然後返回一個新的 RDD。變換函數包括：map，filter，flatMap，groupByKey，reduceByKey，aggregateByKey，pipe和coalesce。
行動（Action）行動操作計算並返回一個新的值。當在一個 RDD 對象上調用行動函數時，會在這一時刻計算全部的數據處理查詢並返回結果值。行動操作包括：reduce，collect，count，first，take，countByKey 以及 foreach。

DataFrame：以RDD為基礎的分散式數據集，與RDD相同，採用惰性機制，只記錄各種轉換的邏輯線路圖(DAG)，支援SQL查詢

2.3 架構設計

Spark運行架構包括集群資源管理器（Cluster Manager）、運行作業任務的工作節點（Worker Node）、每個應用的任務控制節點（Driver）和每個工作節點上負責具體任務的執行進程（Executor）。其中，集群資源管理器可以是Spark自帶的資源管理器，也可以是YARN或Mesos等資源管理框架。

與Hadoop MapReduce計算框架相比，Spark所採用的Executor有兩個優點：

一是利用多執行緒來執行具體的任務（Hadoop MapReduce採用的是進程模型），減少任務的啟動開銷；

二是Executor中有一個BlockManager存儲模組，會將記憶體和磁碟共同作為存儲設備，當需要多輪迭代計算時，可以將中間結果存儲到這個存儲模組里，下次需要時，就可以直接讀該存儲模組里的數據，而不需要讀寫到HDFS等文件系統里，因而有效減少了IO開銷；或者在互動式查詢場景下，預先將錶快取到該存儲系統上，從而可以提高讀寫IO性能。

2.4 Spark運行基本流程

當一個Spark應用被提交時，首先需要為這個應用構建起基本的運行環境，即由任務控制節點（Driver）創建一個SparkContext，由SparkContext負責和資源管理器（Cluster Manager）的通訊以及進行資源的申請、任務的分配和監控等。SparkContext會向資源管理器註冊並申請運行Executor的資源；
資源管理器為Executor分配資源，並啟動Executor進程，Executor運行情況將隨著「心跳」發送到資源管理器上；
SparkContext根據RDD的依賴關係構建DAG圖，DAG圖提交給DAG調度器（DAGScheduler）進行解析，將DAG圖分解成多個「階段」（每個階段都是一個任務集），並且計算出各個階段之間的依賴關係，然後把一個個「任務集」提交給底層的任務調度器（TaskScheduler）進行處理；Executor向SparkContext申請任務，任務調度器將任務分發給Executor運行，同時，SparkContext將應用程式程式碼發放給Executor；
任務在Executor上運行，把執行結果回饋給任務調度器，然後回饋給DAG調度器，運行完畢後寫入數據並釋放所有資源。
3. Hadoop Spark 對比
Spark的中間數據放到記憶體中，對於迭代運算效率更高，API中提供了大量的RDD操作如join，groupby等，而且通過DAG圖可以實現良好的容錯
Spark更適合於迭代運算比較多的ML和DM運算。因為在Spark裡面，有RDD的抽象概念
Spark比Hadoop更通用 - Spark提供的數據集操作類型有很多種，不像Hadoop只提供了Map和Reduce兩種操作。比如map, filter, flatMap, sample, groupByKey, reduceByKey, union, join, cogroup, mapValues, sort,partionBy等多種操作類型，Spark把這些操作稱為Transformations。同時還提供Count, collect, reduce, lookup, save等多種actions操作。 - 這些多種多樣的數據集操作類型，給給開發上層應用的用戶提供了方便。各個處理節點之間的通訊模型不再像Hadoop那樣就是唯一的Data Shuffle一種模式。用戶可以命名，物化，控制中間結果的存儲、分區等。可以說編程模型比Hadoop更靈活。 - 不過由於RDD的特性，Spark不適用那種非同步細粒度更新狀態的應用，例如web服務的存儲或者是增量的web爬蟲和索引。就是對於那種增量修改的應用模型不適合。
容錯性 - 在分散式數據集計算時通過checkpoint來實現容錯，而checkpoint有兩種方式，一個是checkpoint data，一個是logging the updates。用戶可以控制採用哪種方式來實現容錯。
可用性 - Spark通過提供豐富的Scala, Java，Python API及互動式Shell來提高可用性。

兩者都是用MapReduce模型來進行並行計算：

- hadoop的一個作業稱為job，job裡面分為map task和reduce task，每個task都是在自己的進程中運行的，當task結束時，進程也會結束  - hadoop的job只有map和reduce操作，表達能力比較欠缺而且在mr過程中會重複的讀寫hdfs，造成大量的io操作，多個job需要自己管理關係  - spark用戶提交的任務成為application，一個application對應一個SparkContext，app中存在多個job，每觸發一次action操作就會產生一個job  - 這些job可以並行或串列執行，每個job中有多個stage，stage是shuffle過程中DAGSchaduler通過RDD之間的依賴關係劃分job而來的，每個stage裡面有多個task，組成taskset有TaskSchaduler分發到各個executor中執行，executor的生命周期是和app一樣的，即使沒有job運行也是存在的，所以task可以快速啟動讀取記憶體進行計算

4. Ref

Tags: