1、spark技術探秘技術部Spark是什么v Spark是一個基于內存計算的開源的集群(分布式）計算系統(tǒng)v Spark非常小巧玲瓏，由加州伯克利大學AMP實驗室的Matei為主的小團隊所開發(fā)。使用的語言是Scala，項目的core部分的代碼只有63個Scala文件，非常短小精悍。v 由于是基于內存計算，效率要高于擁有Hadoop，Job中間輸出和結果可以保存在內存中，從而不再需要讀寫HDFS，節(jié)省了磁盤IO耗時，號稱性能比Hadoop快100倍。Spark特性v 容錯性：在分布式數據集計算時通過checkpoint來實現容錯，而checkpoint有兩種方式，一個是checkpoint data

2、，一個是logging the updates。用戶可以控制采用哪種方式來實現容錯。v 通用性：Spark通過提供豐富的Scala, Java，Python API及交互式Shell來提高可用性。Spark架構生態(tài)體系Shark(SQL)v shark與hive對比Bagel(Pregel on spark)：v Bagel是基于Spark的輕量級的Pregel(Pregel是Google鼎鼎有名的圖計算框架)的實現。Shark ( Hive on Spark)v Shark是在Spark的框架基礎上提供和Hive一樣的H iveQL命令接口，Shark可以自動在內存中緩存特定的RDD，實現數

3、據重用，進而加快特定數據集的檢索。v Shark通過UDF實現特定的數據分析算法，使得SQL數據查詢和運算分析能結合在一起，最大化RDD的重復使用Sparkstreamingv 流處理系統(tǒng)，實時計算系統(tǒng)v 構建在Spark上處理Stream數據的框架，基本的原理是將Stream數據分成小的時間片斷（幾秒），以類似batch批量處理的方式來處理這小部分數據。Spark Streaming構建在Spark上，一方面是因為Spark的低延遲執(zhí)行引擎（100ms+）可以用于實時計算，另一方面相比基于Record的其它處理框架（如Storm），RDD數據集更容易做高效的容錯處理。此外小批量處理的方式使得

4、它可以同時兼容批量和實時數據處理的邏輯和算法。方便了一些需要歷史數據和實時數據聯合分析的特定應用場合。MLlib(machine learning library)v 機器學習庫：提供高質量的算法,比MapReduce快100倍v 高性能：v Spark擅長迭代計算，這可以使mllib運行的更快，另外，MLlib也包含高效的算法，利用spark的迭代優(yōu)勢，從而產生百倍效果MLlib(machine learning library)v 易部署：如果你有一個hadoop2集群，你可以在沒有任何預裝的情況下運行spark和MLlib。另外，spark也可以運行standalone或EC2或Meso

5、s?？梢宰x取hdfs、hbase或任何一個hadoop的數據源。RDDv RDD(Resilient Distributed Dataset，彈性分布式數據集)v RDD是Spark操縱數據的一個高度抽象，即Spark所操作的數據集都是包裝成RDD來進行操作的，例如Spark可以兼容處理Hadoop的HDFS數據文件，那么這個HDFS數據文件就是包裝成Spark認識的RDD來完成數據抽取和處理的v RDD是Spark的一個核心抽象，Spark的數據操作都是基于RDD來完成。 Map Reducev MR 是Spark可以支撐的運算模式，比傳統(tǒng)的Hadoop MR的性能更好，并且操作集更加豐富。

6、Spark的MR計算引擎的架構圖FPv 函數式編程：functional programmingGraphX(graph)圖計算：alpha版本，已并入spark，暫時不做了解Spark運行模式vStandalone模式v 即獨立模式，自帶完整的服務，可單獨部署到一個集群中，無需依賴任何其他資源管理系統(tǒng)。從一定程度上說，該模式是其他兩種的基礎。v 各個節(jié)點上的資源被抽象成粗粒度的slot，有多少slot就能同時運行多少task。v 不同的是，MapReduce將slot分為map slot和reduce slot，它們分別只能供Map Task和Reduce Task使用，而不能共享，這是Ma

7、pReduce資源利率低效的原因之一，而Spark則更優(yōu)化一些，它不區(qū)分slot類型，只有一種slot，可以供各種類型的Task使用，這種方式可以提高資源利用率，但是不夠靈活，不能為不同類型的Task定制slot資源?？傊?，這兩種方式各有優(yōu)缺點。Spark On Mesos模式vSpark On Mesos模式v 這是很多公司采用的模式，官方推薦這種模式（當然，原因之一是血緣關系）。正是由于Spark開發(fā)之初就考慮到支持Mesos，因此，目前而言，Spark運行在Mesos上會比運行在YARN上更加靈活，更加自然。Spark On YARN模式v Spark On YARN模式v 這是一種最有

8、前景的部署模式。但限于YARN自身的發(fā)展，目前僅支持粗粒度模式（Coarse-grained Mode）。這是由于YARN上的Container資源是不可以動態(tài)伸縮的，一旦Container啟動之后，可使用的資源不能再發(fā)生變化，不過這個已經在YARN計劃（具體參考：/jira/browse/YARN-1197）中了。Spark的適用場景v Spark立足于內存計算，從而不再需要頻繁的讀寫HDFS，這使得Spark能更好的適用于：v 迭代算法，包括大部分機器學習算法Machine Learning和比如PageRank的圖形算法。 v 交互式數據挖

9、掘，用戶大部分情況都會大量重復的使用導入RAM的數據（R、Excel、python） v 需要持續(xù)長時間維護狀態(tài)聚合的流式計算。 Spark數據的存儲v Spark支持多種數據底層存儲，這點比Hadoop支持的數據文件格式廣泛的多。Spark可以兼容HDFS，Hbase,Amazon S3等多種數據集，將這些數據集封裝成RDD進行操作Spark核心概念 RDDv Resilient Distributed Dataset (RDD)彈性分布式數據集v 1、有一個分片列表。就是能被切分，和hadoop一樣的，能夠切分的數據才能并行計算。v 2、有一個函數計算每一個分片， a function f

10、or computing each split 每一個split調用一次map函數v 3、對其他的RDD的依賴列表，也就是rdd演變，進化。依賴分為寬依賴和窄依賴，并不是所有的RDD都有依賴。v 4、可選：key-value型的RDD是根據哈希來分區(qū)的，類似于mapreduce當中的Paritioner接口，控制key分到哪個reduce。v 5、可選：為每一個分片的分配最優(yōu)先計算位置（preferred locations），比如HDFS的block的所在位置應該是優(yōu)先計算的位置。集群配置v Spark-env.sh配置文件v Export JAVA_HOME=/usr/local/jdkv

11、 Export SPARK_MASTER_IP=77v Export SPARK_WORKER_CORES=1v Export SPARK_WORKER_INSTANCES=1v Export WORK_MEMORY=8gv Export SPARK_MASTER_PORT=7077v Export SPARK_JAVA_OPTS=“-verbose:gc XX:-PrintGCDetails XX:+PrintGCTimeStamp”集群配置v Slavesv Xx.xx.xx.2v Xx.xx.xx.3v Xx.xx.xx.4v Xx.xx.xx.5流程示意針對針對R

12、DD的操作的操作Ps：RDD可以從集合直接轉換而來，也可以從現存的任何可以從集合直接轉換而來，也可以從現存的任何Hadoop InputFormat而來，亦或者是而來，亦或者是Hbase等等。等等。加載數據加載數據集集Action觸觸發(fā)執(zhí)行發(fā)執(zhí)行緩存策略v class StorageLevel private(v private var useDisk_ : Boolean,v private var useMemory_ : Boolean,v private var deserialized_ : Boolean,v private var replication_ : Int = 1)緩

13、存策略v val NONE = new StorageLevel(false, false, false)v val DISK_ONLY = new StorageLevel(true, false, false)v val DISK_ONLY_2 = new StorageLevel(true, false, false, 2)v val MEMORY_ONLY = new StorageLevel(false, true, true)Cache默認默認transformation & actionFirst Demov Lines = sc.textFile(“hdfs:/”) /

14、加載進來成為RDDv Errors = lines.filter(_.startsWith(“ERROR”)/transformationv Errors.persist() /緩存RDDv Mysql_errors=errors.fileter(_.contains(“MySQL”).count /actionv http_errors = errors.filter(_.contains(“Http”).countTestCase啟動spark：spark-shellv val rdd= sc.parallelize(List(1,2,3,4,5,6).map(2*_）).filter(_