大數(shù)據(jù)導(dǎo)論-思維、技術(shù)與應(yīng)用-第9章-大數(shù)據(jù)實時處理SPARK

大數(shù)據(jù)導(dǎo)論第九章CONTENTS目錄PART01Spark簡介PART02RDD概念PART03Spark運行架構(gòu)和機制PART04Spark生態(tài)系統(tǒng)PART05Spark編程實踐PART06作業(yè)PART01Spark簡介Spark是加州大學(xué)伯克利分校AMP實驗室開發(fā)的通用內(nèi)存并行計算框架。Spark在2013年6月進入Apache成為孵化項目，8個月后成為Apache頂級項目。Spark以其先進的設(shè)計理念，迅速成為社區(qū)的熱門項目，圍繞著Spark推出了SparkSQL、SparkStreaming、MLlib和GraphX等組件，逐漸形成大數(shù)據(jù)處理一站式解決平臺。Spark簡介Spark優(yōu)勢：1.Spark提供了內(nèi)存計算，減少寫入磁盤的需求，提高了處理效率，更適合交互式分析2.Spark提供了一個統(tǒng)一的框架管理不同來源的數(shù)據(jù)以及支持更多的計算類型3.Spark操作比Hadoop更容易，而且提供的數(shù)據(jù)集類型更加豐富4.Spark基于DAG的調(diào)度機制比Hadoop更優(yōu)越，可以使用更多管道數(shù)據(jù)，控制中間結(jié)果行為

Spark簡介Spark簡介Spark適用場景1.多次操作特定數(shù)據(jù)集的應(yīng)用場合

解釋：Spark是基于內(nèi)存的迭代計算框架，需要反復(fù)操作的次數(shù)越多，所需讀取的數(shù)據(jù)量越大，受益越大；數(shù)據(jù)量小但是計算密集度較大的場合，受益就相對較小2.數(shù)據(jù)量不大，但是需要實時統(tǒng)計分析需求3.由于RDD特性，Spark不適用于增量修改的場景

例如：Web服務(wù)的存儲或者是增量的Web爬蟲和索引PART02RDD概念RDD是Spark提供的最重要的抽象概念，它是一種有容錯機制的特殊數(shù)據(jù)集合，可以分布在集群的節(jié)點上，以函數(shù)式操作集合的方式進行各種并行操作。RDD概念RDD的基本概念1.RDD是一個分布式對象集合，本質(zhì)上是一個只讀的分區(qū)記錄集合。2.每個RDD可以分成多個分區(qū)，每個分區(qū)就是一個數(shù)據(jù)集片段。3.一個RDD的不同分區(qū)可以保存到集群中的不同的節(jié)點上，從而可以在集群中的不同節(jié)點上進行并行計算。RDD概念RDD屬性：是只讀的，不能修改，只能通過轉(zhuǎn)換操作生成新的RDD。是分布式的，可以分布在多臺機器上，進行并行處理。是彈性的，計算過程中內(nèi)存不夠時它會和磁盤進行數(shù)據(jù)交換。是基于內(nèi)存的，可以全部或部分緩存在內(nèi)存中，在多次計算間重用。RDD實質(zhì)上是一種更為通用的迭代并行計算框架，用戶可以顯示控制計算的中間結(jié)果，然后將其自由運用于之后的計算。RDD概念RDD基本操作1.構(gòu)建操作第一類方式從內(nèi)存里構(gòu)造RDD，使用makeRDD和parallelize方法例：valrdd01=sc.makeRDD(List(1,2,3,4,5,6))第二類方式是通過文件系統(tǒng)構(gòu)造RDD例：valrdd:RDD[String]=sc.textFile("file:///D:/sparkdata.txt",1)2.轉(zhuǎn)換操作RDD的轉(zhuǎn)換操作是返回新RDD的操作。轉(zhuǎn)換出來的RDD是惰性求值的，只有在行動操作中用到這些RDD時才會被計算RDD概念

函數(shù)名作用示例結(jié)果map()將函數(shù)應(yīng)用于RDD每個元素，返回值是新的RDDrdd1.map(x=>x+1){2,3,4,4}flatMap()將函數(shù)應(yīng)用于RDD每個元素，將元素數(shù)據(jù)進行拆分，變成迭代器，返回值是新的RDDrdd1.flatmap(x=>x.to(3)){1,2,3,2,3,3,3}filter()函數(shù)會過濾掉不符合條件的元素，返回值是新的RDDrdd1.filter(x=>x!=1){2,3,3}distinct()將RDD里的元素進行去重操作rdd1.distinct()(1,2,3)union()生成包含兩個RDD所有元素的新RDDrdd1.union(rdd2){1,2,3,3,3,4,5}intersection()求出兩個RDD的共同元素ersection(rdd2){3}subtract()將原RDD里和參數(shù)RDD里相同的元素去掉rdd1.subtract(rdd2){1,2}cartesian()求兩個RDD的笛卡兒積rdd1.cartesian(rdd2){(1,3),(1,4),…,(3,5)}RDD概念3.行動操作行動操作用于執(zhí)行計算并按指定的方式輸出結(jié)果。行動操作接受RDD，但是返回非RDD，即輸出一個值或者結(jié)果。在RDD執(zhí)行過程中，真正的計算發(fā)生在行動操作RDD概念函數(shù)名作用示例結(jié)果collect()返回RDD所有元素rdd.collect(){1,2,3,3}count()RDD里元素個數(shù)rdd.count()4countByValue()各元素在RDD中出現(xiàn)次數(shù)rdd.countByValu(){(1,1),(2,1),(3,2})}take(num)從RDD中返回num個元素rdd.take(2){1,2}top(num)從RDD中，按照默認（降序）或者指定的排序返回最前面num個元素rdd.top(2){3,3}reduce()并行整合所有RDD數(shù)據(jù)，例如求和操作rdd.reduce((x,y)=>x+y)9fold(zero)(func)和reduce功能一樣，需要提供初始值rdd.fold(0)((x,y)=>x+y)9foreach(func)對RDD每個元素都使用特定函數(shù)rdd1.foreach(x=>println(x))打印每一個元素saveAsTextFile(path)將數(shù)據(jù)集的元素，以文本的形式保存到文件系統(tǒng)。rdd1.saveAsTextFile(file://home/test)

saveAsSequenceFile(path)將數(shù)據(jù)集的元素，以順序文件格式保存到指定的目錄下。saveAsSequenceFile(hdfs://home/test)

RDD概念RDD血緣關(guān)系：RDD的最重要的特性之一就是血緣關(guān)系（Lineage），血緣關(guān)系描述了一個RDD是如何從父RDD計算得來的。如果某個RDD丟失了，則可以根據(jù)血緣關(guān)系，從父RDD計算得來。RDD概念RDD依賴類型RDD在血緣關(guān)系的依賴分為:窄依賴寬依賴窄依賴是指父RDD的每個分區(qū)都只被子RDD的一個分區(qū)所使用。寬依賴就是指父RDD的分區(qū)被多個子RDD的分區(qū)所依賴。RDD概念階段劃分從Action開始，遇到窄依賴操作，則劃分到同一個執(zhí)行階段；遇到寬依賴操作，則劃分一個新的執(zhí)行階段。后面的Stage需要等待所有的前面的Stage執(zhí)行完之后才可以執(zhí)行RDD概念RDD緩存Spark可以使用persist和cache方法將任意RDD緩存到內(nèi)存、磁盤文件系統(tǒng)中。緩存是容錯的，如果一個RDD分片丟失，可以通過構(gòu)建它的轉(zhuǎn)換自動重構(gòu)。被緩存的RDD被使用時，存取速度會被大大加速。一般Executor內(nèi)存的60%做cache，剩下的40%用來執(zhí)行任務(wù)。cache是persist的特例，將該RDD緩存到內(nèi)存中。而persist可以讓用戶根據(jù)需求指定一個持久化級別級別使用空間CPU時間是否在內(nèi)存是否在磁盤MEMORY_ONLY高低是否MEMORY_ONLY_SER低高是否MEMORY_AND_DISK高中部分部分MEMORY_AND_DISK_SER低高部分部分DISK_ONLY低高否是RDD概念

spark持久化級別的選擇：1.如果RDD可以很好的與默認的存儲級別(MEMORY_ONLY)契合，就不需要做任何修改了。這已經(jīng)是CPU使用效率最高的選項，它使得RDD的操作盡可能的快。2.如果不行，試著使用MEMORY_ONLY_SER并且選擇一個快速序列化庫使得對象在有比較高的空間使用率的情況下，依然可以較快被訪問。RDD概念

spark持久化級別的選擇：3.盡可能不要存儲到硬盤上，除非計算數(shù)據(jù)集函數(shù)的計算量特別大，或它們過濾了大量的數(shù)據(jù)。否則，重新計算一個分區(qū)的速度，和與從硬盤中讀取基本差不多快。4.如果想有快速故障恢復(fù)能力，使用復(fù)制存儲級別。所有的存儲級別都有通過重新計算丟失數(shù)據(jù)恢復(fù)錯誤的容錯機制，但是復(fù)制存儲級別可以讓任務(wù)在RDD上持續(xù)運行，而不需要等待丟失的分區(qū)被重新計算。5.在不使用cachedRDD的時候，及時使用unpersist方法來釋放它PART03Spark運行架構(gòu)和機制Spark運行架構(gòu)和機制Spark總體架構(gòu)Spark運行架構(gòu)包括集群資源管理器（ClusterManager）、多個運行作業(yè)任務(wù)的工作節(jié)點（WorkerNode）、每個應(yīng)用的任務(wù)控制節(jié)點（Driver）和每個工作節(jié)點上負責(zé)具體任務(wù)的執(zhí)行進程（Executor）。Spark運行架構(gòu)和機制

Driver（應(yīng)用控制器）是運行SparkApplication（應(yīng)用）的main()函數(shù)，它會創(chuàng)建SparkContext。SparkContext負責(zé)和ClusterManager（集群管理器）通信，進行資源的申請、任務(wù)的分配和監(jiān)控等。資源管理器負責(zé)申請和管理在WorkerNode（工作節(jié)點）上運行應(yīng)用所需的資源。目前有Spark原生的資源管理器、Mesos資源管理器和HadoopYARN資源管理器三種。Spark運行架構(gòu)和機制

Executor（執(zhí)行進程）是Application運行在WorkerNode上的一個進程，負責(zé)運行Task（任務(wù)），并且負責(zé)將數(shù)據(jù)存在內(nèi)存或者磁盤上，每個Application都有各自獨立的一批Executor。每個Executor則包含了一定數(shù)量的資源來運行分配給它的任務(wù)。每個WorkerNode上的Executor服務(wù)于不同。的Application，它們之間是不可以共享數(shù)據(jù)的Spark運行架構(gòu)和機制SparkApplication的基本概念一個SparkApplication就是用戶編寫的Spark應(yīng)用程序，由一個Driver和若干個Job組成。每個Job都被分隔成多個彼此依賴稱之為Stage的Task。一個Task就是一個工作單元，可以發(fā)送給一個Executor執(zhí)行。Task是用來執(zhí)行應(yīng)用的實際計算工作。Spark運行架構(gòu)和機制Spark運行流程Spark運行架構(gòu)和機制RDD運行過程：構(gòu)建DAG階段DAG調(diào)度階段Task調(diào)度階段執(zhí)行階段Spark生態(tài)圈是伯克利APMLab實驗室打造的，力圖在算法（Algorithms）、機器（Machines）、人（People）之間通過大規(guī)模集成來展現(xiàn)大數(shù)據(jù)應(yīng)用的一個平臺PART04Spark生態(tài)系統(tǒng)Spark生態(tài)系統(tǒng)SparkCoreSparkStreamingSparkSQLSparkMllibSparkGraphXPART05Spark編程實踐Spark編程實踐SparkShell使用1.啟動SparkShell:./spark-shell2.利用本地文件系統(tǒng)的一個文本文件創(chuàng)建一個新RDD3.執(zhí)行動作操作，計算文檔中有多少行

textFile.count()4.執(zhí)行動作操作，獲取文檔中第一行內(nèi)容

textFile.first()5.轉(zhuǎn)換操作會將一個RDD轉(zhuǎn)換成一個新的RDD

varnewRDD=textFile.filter(line=>line.contains(""))

vartextFile=sc.textFile("");Spark編程實踐SparkShell的WordCount實現(xiàn)valfile=sc.textFile("")valcount=file.flatMap(line=>line.split("")).map(word=>(word,1)).reduceByKey(_+_)count.collect()Spark編程實踐importorg.apache.spark.SparkContextimportorg.apache.spark.SparkContext._importorg.apache.spark.SparkConfobjectSimpleApp{defmain(args:Array[String]){vallogFile="file:///usr/local/spark/README.md"http://Shouldbesomefileonyoursystemvalconf=newSparkConf().setAppName("SimpleApplication")valsc=newSparkContext(conf)vallogData=sc.textFile(logFile,2).cache()valnumAs=logData.filter(line=>line.contains("a")).count()valnumBs=logData.filter(line=>line.contains("b")).count()println("Lineswitha:%s,Lineswithb:%s".format(numAs,numBs))}}編寫Scala應(yīng)用程序Spark編程實踐編寫Java應(yīng)用程序importorg.apache.spark.api.java.*;importorg.apache.spark.api.java.function.Function;publicclassSimpleApp{publicstaticvoidmain(String[]args){StringlogFile="file:///usr/local/spark/README.md";//ShouldbesomefileonyoursystemJavaSparkContextsc=newJavaSparkContext("local","SimpleApp","file:///usr/local/spark/",newString[]{"target/simple-project-1.0.jar"});JavaRDD<String>logData=sc.textFile(logFile).cache();longnumAs=logData.filter(newFunction<String,Boolean>(){publicBooleancall(Strings){returns.contains("a");}}).count();longnumBs=logData.filter(newFunction<String,Boolean>(){publicBooleancall(Strings){returns.contains("b");}}).count();System.out.println("Lineswitha:"+numAs+",lineswithb:"+numBs);}} PART06作業(yè)

作業(yè)作業(yè)：1.Hadoop的主要缺陷是什么？Spark的主要優(yōu)勢是什么？2.Spark的主要使用場景有哪些？3.什么是RDD？它有哪些主要屬性？產(chǎn)生RDD的主要原因是什么？4.RDD有哪幾類操作？各自的主要作用是什么？5.什么是Spark的惰性計算？有什么優(yōu)勢？6.什么是RDD的血緣關(guān)系？Spark保存RDD血緣關(guān)系的用途是什么？7.什么是DAG？請舉例說明。8.RDD的依賴關(guān)系有寬依賴和窄依賴兩大類，請解釋什么是寬依賴，什么是窄依賴？把依賴關(guān)系分解為寬依賴和窄依賴的目的是什么？作業(yè)作業(yè)：9.圖9.13展示的轉(zhuǎn)換中，哪些是寬依賴？哪些是窄依賴？并解釋原因作業(yè)作業(yè)：10.Spark是如何把DAG劃分成多個階段的？劃分成多個階段的依據(jù)是什么？劃分成多個階段的目的是什么？11.請對圖9.14DAG進行階段劃分，并解釋階段劃分的原因。作業(yè)

12.RDD的緩存的主要目的是什么？有哪幾類緩存級別？13.請畫圖描述Spark的運行架構(gòu)，并解釋每一個模塊的主要作用。14.請畫圖描述Spark應(yīng)用的運行流程，并簡單描述每一步的任務(wù)。15.假定rdd1={2，3，3，5},rdd2={3,4,5}，請寫出下面轉(zhuǎn)換操作的結(jié)果函數(shù)名作用示例結(jié)果map()將函數(shù)應(yīng)用于RDD每個元素，返回值是新的R