大數(shù)據(jù)之spark學(xué)習(xí)

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

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

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

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

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

6、Spark的MR計(jì)算引擎的架構(gòu)圖FPv 函數(shù)式編程：functional programmingGraphX(graph)圖計(jì)算：alpha版本，已并入spark，暫時(shí)不做了解Spark運(yùn)行模式vStandalone模式v 即獨(dú)立模式，自帶完整的服務(wù)，可單獨(dú)部署到一個(gè)集群中，無需依賴任何其他資源管理系統(tǒng)。從一定程度上說，該模式是其他兩種的基礎(chǔ)。v 各個(gè)節(jié)點(diǎn)上的資源被抽象成粗粒度的slot，有多少slot就能同時(shí)運(yùn)行多少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)缺點(diǎn)。Spark On Mesos模式vSpark On Mesos模式v 這是很多公司采用的模式，官方推薦這種模式（當(dāng)然，原因之一是血緣關(guān)系）。正是由于Spark開發(fā)之初就考慮到支持Mesos，因此，目前而言，Spark運(yùn)行在Mesos上會(huì)比運(yùn)行在YARN上更加靈活，更加自然。Spark On YARN模式v Spark On YARN模式v 這是一種最有

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

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

10、or computing each split 每一個(gè)split調(diào)用一次map函數(shù)v 3、對(duì)其他的RDD的依賴列表，也就是rdd演變，進(jìn)化。依賴分為寬依賴和窄依賴，并不是所有的RDD都有依賴。v 4、可選：key-value型的RDD是根據(jù)哈希來分區(qū)的，類似于mapreduce當(dāng)中的Paritioner接口，控制key分到哪個(gè)reduce。v 5、可選：為每一個(gè)分片的分配最優(yōu)先計(jì)算位置（preferred locations），比如HDFS的block的所在位置應(yīng)該是優(yōu)先計(jì)算的位置。集群配置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流程示意針對(duì)針對(duì)R

12、DD的操作的操作Ps：RDD可以從集合直接轉(zhuǎn)換而來，也可以從現(xiàn)存的任何可以從集合直接轉(zhuǎn)換而來，也可以從現(xiàn)存的任何Hadoop InputFormat而來，亦或者是而來，亦或者是Hbase等等。等等。加載數(shù)據(jù)加載數(shù)據(jù)集集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默認(rèn)默認(rèn)transformation & actionFirst Demov Lines = sc.textFile(“hdfs:/”) /

14、加載進(jìn)來成為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啟動(dòng)spark：spark-shellv val rdd= sc.parallelize(List(1,2,3,4,5,6).map(2*_）).filter(_