大數據項目實戰必備技能之Spark
Spark 原理簡述
Spark 是使用 scala 實現的基於內存計算的大數據開源集群計算環境.提供了 java,scala, python,R 等語言的調用接口.
Hadoop 和 Spark 的關系
Google 在 2003 年和 2004 年先後發表了 Google 文件系統 GFS 和 MapReduce 編程模型兩篇文章,. 基於這兩篇開源文檔,06 年 Nutch 項目子項目之一的 Hadoop 實現了兩個強有力的開源產品:HDFS 和 MapReduce. Hadoop 成為了典型的大數據批量處理架構,由 HDFS 負責靜態數據的存儲,並通過 MapReduce 將計算邏輯分配到各數據節點進行數據計算和價值發現.之後以 HDFS 和 MapReduce 為基礎建立了很多項目,形成了 Hadoop 生態圈.
而 Spark 則是UC Berkeley AMP lab (加州大學伯克利分校AMP實驗室)所開源的類Hadoop MapReduce的通用並行框架, 專門用於大數據量下的叠代式計算.是為了跟 Hadoop 配合而開發出來的,不是為了取代 Hadoop, Spark 運算比 Hadoop 的 MapReduce 框架快的原因是因為 Hadoop 在一次 MapReduce 運算之後,會將數據的運算結果從內存寫入到磁盤中,第二次 Mapredue 運算時在從磁盤中讀取數據,所以其瓶頸在2次運算間的多余 IO 消耗. Spark 則是將數據一直緩存在內存中,直到計算得到最後的結果,再將結果寫入到磁盤,所以多次運算的情況下, Spark 是比較快的. 其優化了叠代式工作負載[^demo_zongshu].
具體區別如下:
伯克利大學將 Spark 的整個生態系統成為 伯克利數據分析棧(BDAS),在核心框架 Spark 的基礎上,主要提供四個範疇的計算框架:
-
Spark SQL: 提供了類 SQL 的查詢,返回 Spark-DataFrame 的數據結構(類似 Hive)
-
Spark Streaming: 流式計算,主要用於處理線上實時時序數據(類似 storm)
-
MLlib: 提供機器學習的各種模型和調優
- GraphX: 提供基於圖的算法,如 PageRank
Spark 的主要特點還包括:
-
(1)提供 Cache 機制來支持需要反復叠代計算或者多次數據共享,減少數據讀取的 IO 開銷;
-
(2)提供了一套支持 DAG 圖的分布式並行計算的編程框架,減少多次計算之間中間結果寫到 Hdfs 的開銷;
- (3)使用多線程池模型減少 Task 啟動開稍, shuffle 過程中避免不必要的 sort 操作並減少磁盤 IO 操作。(Hadoop 的 Map 和 reduce 之間的 shuffle 需要 sort)
Spark 系統架構
首先明確相關術語[^demo_shuyu]:
-
應用程序(Application): 基於Spark的用戶程序，包含了一個Driver Program 和集群中多個的Executor；
-
驅動(Driver): 運行Application的main()函數並且創建SparkContext;
-
執行單元(Executor): 是為某Application運行在Worker Node上的一個進程，該進程負責運行Task，並且負責將數據存在內存或者磁盤上，每個Application都有各自獨立的Executors;
-
集群管理程序(Cluster Manager): 在集群上獲取資源的外部服務(例如：Local、Standalone、Mesos或Yarn等集群管理系統)；
- 操作(Operation): 作用於RDD的各種操作分為Transformation和Action.
整個 Spark 集群中,分為 Master 節點與 worker 節點,,其中 Master 節點上常駐 Master 守護進程和 Driver 進程, Master 負責將串行任務變成可並行執行的任務集Tasks, 同時還負責出錯問題處理等,而 Worker 節點上常駐 Worker 守護進程, Master 節點與 Worker 節點分工不同, Master 負載管理全部的 Worker 節點,而 Worker 節點負責執行任務.
Driver 的功能是創建 SparkContext, 負責執行用戶寫的 Application 的 main 函數進程,Application 就是用戶寫的程序.
Spark 支持不同的運行模式,包括Local, Standalone,Mesoses,Yarn 模式.不同的模式可能會將 Driver 調度到不同的節點上執行.集群管理模式裏, local 一般用於本地調試.
每個 Worker 上存在一個或多個 Executor 進程,該對象擁有一個線程池,每個線程負責一個 Task 任務的執行.根據 Executor 上 CPU-core 的數量,其每個時間可以並行多個 跟 core 一樣數量的 Task[^demopingtai].Task 任務即為具體執行的 Spark 程序的任務.
spark 運行原理
底層詳細細節介紹:
我們使用spark-submit提交一個Spark作業之後，這個作業就會啟動一個對應的Driver進程。根據你使用的部署模式（deploy-mode）不同，Driver進程可能在本地啟動，也可能在集群中某個工作節點上啟動。而Driver進程要做的第一件事情，就是向集群管理器（可以是Spark Standalone集群，也可以是其他的資源管理集群，美團?大眾點評使用的是YARN作為資源管理集群）申請運行Spark作業需要使用的資源，這裏的資源指的就是Executor進程。YARN集群管理器會根據我們為Spark作業設置的資源參數，在各個工作節點上，啟動一定數量的Executor進程，每個Executor進程都占有一定數量的內存和CPU core。
在申請到了作業執行所需的資源之後，Driver進程就會開始調度和執行我們編寫的作業代碼了。Driver進程會將我們編寫的Spark作業代碼分拆為多個stage，每個stage執行一部分代碼片段，並為每個stage創建一批Task，然後將這些Task分配到各個Executor進程中執行。Task是最小的計算單元，負責執行一模一樣的計算邏輯（也就是我們自己編寫的某個代碼片段），只是每個Task處理的數據不同而已。一個stage的所有Task都執行完畢之後，會在各個節點本地的磁盤文件中寫入計算中間結果，然後Driver就會調度運行下一個stage。下一個stage的Task的輸入數據就是上一個stage輸出的中間結果。如此循環往復，直到將我們自己編寫的代碼邏輯全部執行完，並且計算完所有的數據，得到我們想要的結果為止。
Spark是根據shuffle類算子來進行stage的劃分。如果我們的代碼中執行了某個shuffle類算子（比如reduceByKey、join等），那麽就會在該算子處，劃分出一個stage界限來。可以大致理解為，shuffle算子執行之前的代碼會被劃分為一個stage，shuffle算子執行以及之後的代碼會被劃分為下一個stage。因此一個stage剛開始執行的時候，它的每個Task可能都會從上一個stage的Task所在的節點，去通過網絡傳輸拉取需要自己處理的所有key，然後對拉取到的所有相同的key使用我們自己編寫的算子函數執行聚合操作（比如reduceByKey()算子接收的函數）。這個過程就是shuffle。
當我們在代碼中執行了cache/persist等持久化操作時，根據我們選擇的持久化級別的不同，每個Task計算出來的數據也會保存到Executor進程的內存或者所在節點的磁盤文件中。
因此Executor的內存主要分為三塊：第一塊是讓Task執行我們自己編寫的代碼時使用，默認是占Executor總內存的20%；第二塊是讓Task通過shuffle過程拉取了上一個stage的Task的輸出後，進行聚合等操作時使用，默認也是占Executor總內存的20%；第三塊是讓RDD持久化時使用，默認占Executor總內存的60%。
Task的執行速度是跟每個Executor進程的CPU core數量有直接關系的。一個CPU core同一時間只能執行一個線程。而每個Executor進程上分配到的多個Task，都是以每個Task一條線程的方式，多線程並發運行的。如果CPU core數量比較充足，而且分配到的Task數量比較合理，那麽通常來說，可以比較快速和高效地執行完這些Task線程。
以上就是Spark作業的基本運行原理的說明.
在實際編程中,我們不需關心以上調度細節.只需使用 Spark 提供的指定語言的編程接口調用相應的 API 即可.
在 Spark API 中, 一個 應用(Application) 對應一個 SparkContext 的實例。一個 應用 可以用於單個 Job，或者分開的多個 Job 的 session，或者響應請求的長時間生存的服務器。與 MapReduce 不同的是，一個 應用 的進程（我們稱之為 Executor)，會一直在集群上運行，即使當時沒有 Job 在上面運行。
而調用一個Spark內部的 Action 會產生一個 Spark job 來完成它。 為了確定這些job實際的內容，Spark 檢查 RDD 的DAG再計算出執行 plan 。這個 plan 以最遠端的 RDD 為起點（最遠端指的是對外沒有依賴的 RDD 或者 數據已經緩存下來的 RDD），產生結果 RDD 的 Action 為結束 。並根據是否發生 shuffle 劃分 DAG 的 stage.
// parameterval appName = "RetailLocAdjust"val master = "local" // 選擇模式val conf = new SparkConf().setMaster(master).setAppName(appName)// 啟動一個 SparkContext Applicationval sc = new SparkContext(conf)val rdd = sc.textFile("path/...")
要啟動 Spark 運行程序主要有兩種方式:一種是使用 spark-submit 將腳本文件提交,一種是打開 Spark 跟某種特定語言的解釋器,如:
-
spark-shell: 啟動了 Spark 的 scala 解釋器.
-
pyspark: 啟動了 Spark 的 python 解釋器.
- sparkR: 啟動了 Spark 的 R 解釋器.
(以上解釋器位於spark 的 bin 目錄下)
下面整理了一些比較適合新手入門Spark的技術書籍，對Spark技術感興趣想要學習的同學可以做個參考。
1.大數據技術叢書:Spark快速數據處理
Spark快速數據處理
內容淺顯易懂，極其適合入門。從實用角度系統講解Spark的數據處理工具及使用方法，手把手教你充分利用Spark提供的各種功能，快速編寫高效分布式程序。
2.Spark大數據處理：技術、應用與性能優化
Spark大數據處理：技術、應用與性能優化
作者結合自己在微軟和IBM的實踐經驗和對Spark源代碼的研究撰寫而成。首先從技術層面講解了Spark的體系結構、工作機制、安裝與部署、開發環境搭建、計算模型、Benchmark、BDAS等內容；然後從應用角度講解了一些簡單的、有代表性的案例；最後對Spark的性能優化進行了探討。
3.大數據Spark企業級實戰
大數據Spark企業級實戰
完全從企業處理大數據業務場景的角度出發，完全基於實戰代碼來組織內容，從零起步，不許任何基礎，完全無痛地掌握Spark大數據處理實戰技術。
4.Spark大數據處理技術
Spark大數據處理技術
首部全面介紹Spark及Spark生態圈相關技術的技術書籍俯覽未來大局，不失精細剖析，呈現一個現代大數據框架的架構原理和實現細節透徹講解Spark原理和架構，以及部署模式、調度框架、存儲管理及應用監控等重要模塊Spark生態圈深度檢閱：SQL處理Shark和Spark SQL、流式處理Spark Streaming、圖計算Graphx及內存文件系統Tachyon。
今天就給大家分享到這裏，希望對大家有所幫助，希望大家多多關註哦，想了解大數據，學習大數據的小夥伴們，需要大數據資料的可以加我的群725197860，歡迎大家進群討論
大數據項目實戰必備技能之Spark