Spark各個(gè)知識(shí)點(diǎn)總結(jié)講解.ppt

1、Spark簡(jiǎn)介,Spark是什么,Spark是一個(gè)快速且通用的集群計(jì)算平臺(tái)。 集群計(jì)算 把一臺(tái)電腦無(wú)法解決的問(wèn)題，放到多臺(tái)電腦組成的集群上進(jìn)行解決，這就是集群計(jì)算。,Spark的特點(diǎn),Spark是快速的 很多任務(wù)能夠秒級(jí)完成，對(duì)于一些特定的工作，Spark比Mapreduce快10-20倍。 Spark擴(kuò)充了流行的Mapreduce計(jì)算模型，使Spark更高效地支持更多類(lèi)型的計(jì)算，包括交互式查詢，和流處理。 速度快的另一個(gè)主要原因就是，能夠在內(nèi)存中計(jì)算。,Spark的特點(diǎn),Spark是通用的 Spark的設(shè)計(jì)，容納了之前很多獨(dú)立的，分布式系統(tǒng)所擁有的功能。獨(dú)立的分布式系統(tǒng)包括：批處理，迭代式計(jì)

2、算，交互查詢和流處理等。 并且，由之前需要維護(hù)不同的集群，到現(xiàn)在只需要維護(hù)一個(gè)Spark集群。,Spark的特點(diǎn),Spark是高度開(kāi)放的 Spark提供了Python，Java，Scala，SQL的API和豐富的內(nèi)置庫(kù)。 同時(shí)，Spark和其它的大數(shù)據(jù)工具整合的很好。尤其，Spark能夠運(yùn)行在Hadoop集群上面，能夠訪問(wèn)Hadoop數(shù)據(jù)。,Spark的組件,Spark包括多個(gè)緊密集成的組件。,Spark的組件,緊密集成的優(yōu)點(diǎn)： 如果Spark底層優(yōu)化了，那么基于Spark底層的組件，也得到了相應(yīng)的優(yōu)化。例如，Spark底層增加了一個(gè)優(yōu)化算法，那么Spark的SQL和機(jī)器學(xué)習(xí)包也會(huì)自動(dòng)的優(yōu)化。

3、 緊密集成，節(jié)省了各個(gè)組件組合使用時(shí)的部署，測(cè)試等時(shí)間。 當(dāng)向Spark增加新的組件時(shí)，其它的組件，可以立刻享用新組件的功能。 無(wú)縫連接不同的處理模型。,Spark的組件,Spark Core： 包含Spark的基本功能，包含任務(wù)調(diào)度，內(nèi)存管理，容錯(cuò)機(jī)制等。 Spark Core內(nèi)部定義了RDDs(resilient distributed datasets，彈性分布式數(shù)據(jù)集)。RDDs代表橫跨很多工作節(jié)點(diǎn)的數(shù)據(jù)集合，RDDs可以被并行的處理。 Spark Core提供了很多APIs來(lái)創(chuàng)建和操作這些集合(RDDs)。,Spark的組件,Spark SQL： 是Spark處理結(jié)構(gòu)化數(shù)據(jù)的庫(kù)。它支

4、持通過(guò)SQL查詢數(shù)據(jù)，就像HQL(Hive SQL)一樣，并且支持很多數(shù)據(jù)源，像Hive表，JSON等。Spark SQL是在Spark 1.0版本中新加的。 Shark是一種較老的基于Spark的SQL項(xiàng)目，它是基于Hive修改的，它現(xiàn)在已經(jīng)被Spark SQL替代了。,Spark的組件,Spark Streaming: 是實(shí)時(shí)數(shù)據(jù)流處理組件，類(lèi)似Storm。 Spark Streaming提供了API來(lái)操作實(shí)時(shí)流數(shù)據(jù)。,Spark的組件,MLlib: Spark有一個(gè)包含通用機(jī)器學(xué)習(xí)功能的包，就是MLlib(machine learning lib)。 MLlib包含了分類(lèi)，聚類(lèi)，回歸，協(xié)

5、同過(guò)濾算法，還包括模型評(píng)估，和數(shù)據(jù)導(dǎo)入。 它還提供了一些低級(jí)的機(jī)器學(xué)習(xí)原語(yǔ)，包括通用梯度下降優(yōu)化算法。 MLlib提供的上面這些方法，都支持集群上的橫向擴(kuò)展。,Spark的組件,Graphx: 是處理圖的庫(kù)(例如，社交網(wǎng)絡(luò)圖)，并進(jìn)行圖的并行計(jì)算。就像Spark Streaming和Spark SQL一樣，Graphx也繼承了Spark RDD API，同時(shí)允許創(chuàng)建有向圖。 Graphx提供了各種圖的操作，例如subgraph和mapVertices，也包含了常用的圖算法，例如PangeRank等。,Spark的組件,Cluster Managers: Cluster Managers就是集群

6、管理。Spark能夠運(yùn)行在很多cluster managers上面，包括Hadoop YARN，Apache Mesos和Spark自帶的單獨(dú)調(diào)度器。 如果你把Spark安裝在了裸機(jī)上面，單獨(dú)調(diào)度器能夠提供簡(jiǎn)單的方式，讓你開(kāi)始Spark之旅。 如果你已經(jīng)有了Hadoop Yarn或者M(jìn)esos集群，那么，Spark對(duì)這些集群管理工具的支持，使你的Spark應(yīng)用程序能夠在這些集群上面運(yùn)行。,Spark的歷史,Spark誕生于2009年，那時(shí)候它是，加州大學(xué)伯克利分校RAD實(shí)驗(yàn)室的一個(gè)研究項(xiàng)目，后來(lái)到了AMP實(shí)驗(yàn)室。 Spark最初是基于Hadoop Mapreduce的，后來(lái)發(fā)現(xiàn)Mapreduc

7、e在迭代式計(jì)算和交互式上是低效的。因此Spark進(jìn)行了改進(jìn)，引入了內(nèi)存存儲(chǔ)和高容錯(cuò)機(jī)制。 關(guān)于Spark的研究論文在學(xué)術(shù)會(huì)議上發(fā)表，并且在它被創(chuàng)建的2009年不久之后，對(duì)于一些特定的工作，Spark比Mapreduce快10-20倍。 2010年3月份Spark開(kāi)源。 2011年，AMP實(shí)驗(yàn)室開(kāi)始在Spark上面開(kāi)發(fā)高級(jí)組件，像Shark(Hive on Spark)，Spark Streaming。 2013年轉(zhuǎn)移到了Apache下，現(xiàn)在已經(jīng)是頂級(jí)項(xiàng)目了。 2014年5月份Spark1.0發(fā)布。,Spark運(yùn)行環(huán)境,Spark 是Scala寫(xiě)的, 運(yùn)行在JVM上。所以運(yùn)行環(huán)境是Java6或者

8、以上。 如果想要使用 Python API，需要安裝Python 解釋器2.6版本或者以上。 目前Spark(1.2.0版本) 與Python 3不兼容。,Spark下載,下載地址：/downloads.html，選擇Pre-built for Hadoop 2.4 and later 這個(gè)包，點(diǎn)擊直接下載，這會(huì)下載一個(gè)spark-1.2.0-bin-hadoop2.4.tgz的壓縮包 搭建Spark不需要Hadoop，如果你有hadoop集群或者h(yuǎn)dfs，你可以下載相應(yīng)的版本。 解壓：tar -zxvf spark-1.2.0-bin-hadoop

9、2.4.tgz,Spark目錄,README.md 開(kāi)始Spark之旅的簡(jiǎn)單介紹。 bin 包含用來(lái)和Spark交互的可執(zhí)行文件，如Spark shell。 core, streaming, python, 包含主要組件的源代碼。 examples 包含一些有用的單機(jī)Spark job。 你可以研究和運(yùn)行這些例子，來(lái)學(xué)習(xí)Spark API。,Spark的Shells,Spark的shell使你能夠處理分布在集群上的數(shù)據(jù)(這些數(shù)據(jù)可以是分布在硬盤(pán)上或者內(nèi)存中)。 Spark可以把數(shù)據(jù)加載到工作節(jié)點(diǎn)的內(nèi)存中，因此，許多分布式處理(甚至是分布式的1T數(shù)據(jù)的處理)都可以在幾秒內(nèi)完成。 上面的特性，使迭

10、代式計(jì)算，實(shí)時(shí)查詢、分析一般能夠在shells中完成。Spark提供了Python shells和 Scala shells。,Spark的Shells,打開(kāi)Spark的Python Shell： 到Spark目錄，Spark的Python Shell也叫做PySpark Shell bin/pyspark 打開(kāi)PySpark Shell之后的界面,Spark的Shells,打開(kāi)Spark的Scala Shell： 到Spark目錄 bin/pysparkbin/spark-shell打開(kāi)Scala版本的shell 打開(kāi)之后的界面,Spark的Shells,例子： scala val line

11、s = sc.textFile(././testfile/helloSpark) / 創(chuàng)建一個(gè)叫l(wèi)ines的RDD lines: org.apache.spark.rdd.RDDString = ././testfile/helloSpark MappedRDD1 at textFile at :12 scala lines.count() / 對(duì)這個(gè)RDD中的行數(shù)進(jìn)行計(jì)數(shù) res0: Long = 2 scala lines.first() / 文件中的第一行 res1: String = hello spark 修改日志級(jí)別：conf/perties log4j.root

12、Category=WARN, console,Spark的核心概念,Driver program： 包含程序的main()方法，RDDs的定義和操作。(在上面的例子中，driver program就是Spark Shell它本身了) 它管理很多節(jié)點(diǎn)，我們稱作executors。 count()操作解釋(每個(gè)executor計(jì)算文件的一部分，最后合并)。,Spark的核心概念,SparkContext: Driver programs 通過(guò)一個(gè) SparkContext 對(duì)象訪問(wèn) Spark，SparkContext 對(duì)象代表和一個(gè)集群的連接。 在Shell中SparkContext 自動(dòng)創(chuàng)建好

13、了，就是sc， 例子： sc 變量 sc ,Spark的核心概念,RDDs： 在Spark中，我們通過(guò)分布式集合(distributed collections，也就是RDDs)來(lái)進(jìn)行計(jì)算，這些分布式集合，并行的分布在整個(gè)集群中。 RDDs 是 Spark分發(fā)數(shù)據(jù)和計(jì)算的基礎(chǔ)抽象類(lèi)。 用SparkContext創(chuàng)建RDDs 上面例子中使用sc.textFile()創(chuàng)建了一個(gè)RDD，叫l(wèi)ines，它是從我們的本機(jī)文本文件中創(chuàng)建的，這個(gè)RDD代表了一個(gè)文本文件的每一行。我們可以在RDD上面進(jìn)行各種并行化的操作，例如計(jì)算數(shù)據(jù)集中元素的個(gè)數(shù)或者打印出第一行。,Spark的核心概念,向Spark傳遞函數(shù)

14、： 向Spark傳遞函數(shù)是Spark的一個(gè)常用功能，許多Spark API是圍繞它展開(kāi)的。 例子：filtering scala val lines = sc.textFile(././testfile/helloSpark) lines: spark.RDDString = MappedRDD. scala val worldLines = lines.filter(line = line.contains(world) pythonLines: spark.RDDString = FilteredRDD. scala worldLines .collect(),Spark的核心概念,向Sp

15、ark傳遞函數(shù)： 上面例子中的=語(yǔ)法是 Scala中定義函數(shù)的便捷方法。你也可以先定義函數(shù)再引用: 例子： def hasWorld(line:String) : Boolean= line.contains(world) worldLines = lines.filter(hasWorld) 像filter 這樣的基于函數(shù)的操作，也是在集群上并行執(zhí)行的。,Spark的核心概念,向Spark傳遞函數(shù)： 需要注意的地方： 如果你傳遞的函數(shù)是一個(gè)對(duì)象的成員，或者包含一個(gè)對(duì)象中字段的引用(例如self.field)，Spark會(huì)把整個(gè)對(duì)象都發(fā)送到工作節(jié)點(diǎn)上，這樣會(huì)比僅僅發(fā)送你關(guān)心的信息要大很多，而且

16、有時(shí)候會(huì)帶來(lái)一些奇怪的問(wèn)題。 傳送信息太多解決方法：我們可以把關(guān)心的字段抽取出來(lái)，只傳遞關(guān)心的字段。 奇怪問(wèn)題的避免：序列化包含函數(shù)的對(duì)象，函數(shù)和函數(shù)中引用的數(shù)據(jù)都需要序列化(實(shí)現(xiàn)Java的Serializable interface)。 如果Scala中出現(xiàn)NotSerializableException，一般情況下，都是因?yàn)闆](méi)序列化。,RDDs介紹,RDDs介紹 RDDs的創(chuàng)建方法 Scala的基礎(chǔ)知識(shí),RDDs介紹,RDDs Resilient distributed datasets(彈性分布式數(shù)據(jù)集，簡(jiǎn)寫(xiě)RDDs)。 一個(gè)RDD就是一個(gè)不可改變的分布式集合對(duì)象，內(nèi)部由許多partit

17、ions(分片)組成，每個(gè)partition都包括一部分?jǐn)?shù)據(jù)，這些partitions可以在集群的不同節(jié)點(diǎn)上計(jì)算 Partitions是Spark中的并行處理的單元。Spark順序的，并行的處理partitions。 RDDs 是 Spark的分發(fā)數(shù)據(jù)和計(jì)算的基礎(chǔ)抽象類(lèi)，是Spark的核心概念。 RDD可以包含 Python, Java, 或者 Scala中的任何數(shù)據(jù)類(lèi)型，包括用戶自定義的類(lèi)。 在Spark中，所有的計(jì)算都是通過(guò)RDDs的創(chuàng)建，轉(zhuǎn)換，操作完成的。 RDD具有l(wèi)ineage graph(血統(tǒng)關(guān)系圖)。,RDDs的創(chuàng)建方法,Driver program中創(chuàng)建RDDs： 把一個(gè)存在的

18、集合傳給SparkContexts parallelize()方法。這種方法，一般只適用于學(xué)習(xí)時(shí)。 例子： val lines = sc.parallelize(List(spark, bigdatastudy) val rdd = sc.parallelize(Array(1, 2, 2, 4), 4) . 注意一下RDD的類(lèi)型 第一個(gè)參數(shù)是：待并行化處理的集合 第二個(gè)參數(shù)是：分區(qū)個(gè)數(shù),RDDs的創(chuàng)建方法,加載外部數(shù)據(jù)集: 例子：使用textFile() 加載 val rddText= sc.textFile(././testfile/helloSpark) val rddHdfs = sc

19、.textFile(hdfs:/some/path.txt),Scala的基礎(chǔ)知識(shí),Scala的變量聲明 在Scala中創(chuàng)建變量的時(shí)候，必須使用val或者var Val，變量值不可修改，一旦分配不能重新指向別的值 Var，分配后，可以指向類(lèi)型相同的值。,Scala的基礎(chǔ)知識(shí),Scala的變量聲明 val lines= sc.textFile(././testfile/helloSpark) lines= sc.textFile(././testfile/helloSpark2) . : error: reassignment to val var lines2= sc.textFile(./.

20、/testfile/helloSpark) lines2= sc.textFile(././testfile/helloSpark2) 可以重新聲明變量 val lines= sc.textFile(././testfile/helloSpark2),Scala的基礎(chǔ)知識(shí),Scala的匿名函數(shù) 像Python的lambda 函數(shù) lines.filter(line = line.contains(world) . 我們定義一個(gè)匿名函數(shù)，接收一個(gè)參數(shù)line，并使用line這個(gè)String類(lèi)型變量上的contains方法，并且返回結(jié)果。 line 的類(lèi)型不需指定，能夠推斷出來(lái),Scala的基礎(chǔ)知

21、識(shí),Scala程序員就是不喜歡多寫(xiě)代碼。 Scala允許我們用下劃線_來(lái)代表匿名函數(shù)中的參數(shù)。 lines.filter(_.contains(world) .,Scala的基礎(chǔ)知識(shí),類(lèi)型推斷 def hasWorld(line:String) : Boolean=line.contains(world) worldLines = lines.filter(hasWorld) Scala中定義函數(shù)用def，參數(shù)指定類(lèi)型String，因?yàn)楹竺娴腸ontains方法就是用的String中的Contains方法。 函數(shù)返回的類(lèi)型，可以不必指定，因?yàn)橥ㄟ^(guò)類(lèi)型推斷，能夠推出來(lái)。,Scala的基礎(chǔ)知識(shí),類(lèi)

22、型推斷 指定返回類(lèi)型： 返回的類(lèi)型比較復(fù)雜，Scala可能推斷不出來(lái)。 程序更易讀。,Transformations,Transformations介紹 逐元素transformations 集合運(yùn)算,Transformations介紹,Transformations(轉(zhuǎn)換)： 從之前的RDD構(gòu)建一個(gè)新的RDD，像map() 和 filter()。,Transformations介紹,Transformations的特點(diǎn)： Transformations返回一個(gè)嶄新的RDD， filter() 操作返回一個(gè)指針，指向一個(gè)嶄新的RDD，原RDD不受影響，能夠在后面重復(fù)利用。,逐元素transfo

23、rmations,許多的transformations是逐元素的，也就是每次轉(zhuǎn)變一個(gè)元素。 兩個(gè)最常用的transformations：map() and filter() map() transformation，接收一個(gè)函數(shù)，把這個(gè)函數(shù)應(yīng)用到RDD的每一個(gè)元素，并返一個(gè)函數(shù)作用后的新的RDD。 filter() transformation，接收一個(gè)函數(shù)，返回只包含滿足filter()函數(shù)的元素的新RDD。 輸入RDD與輸出RDD可以是不同的類(lèi)型，例如input RDDString ，output RDDDouble,逐元素transformations,map() 例子- 對(duì)RDD中元素

24、求平方 val input = sc.parallelize(List(1, 2, 3, 4) val result = input.map(x = x * x) println(result.collect().mkString(,),逐元素transformations,flatMap() 對(duì)每個(gè)輸入元素，輸出多個(gè)輸出元素。 flat壓扁的意思，將RDD中元素壓扁后返回一個(gè)新的RDD。 例子- flatMap() ,把一行字分割成多個(gè)元素 val lines = sc.parallelize(List(hello world, hi) val words = lines.flatMap(l

25、ine = line.split( ) words.first() / returns hello,逐元素transformations,flatMap(),集合運(yùn)算,RDDs支持?jǐn)?shù)學(xué)集合的計(jì)算，例如并集，交集計(jì)算。注意：進(jìn)行計(jì)算的RDDs應(yīng)該是相同類(lèi)型。 money-monkey,集合運(yùn)算,distinct()是很耗時(shí)的操作，因?yàn)樗枰ㄟ^(guò)網(wǎng)絡(luò)，shuffle所有的數(shù)據(jù)，以保證元素不重復(fù)。 一般情況下，我們不用distinct()。 union(other) 會(huì)包含重復(fù)的元素。 intersection(other)求交集。耗時(shí)操作，因?yàn)樾枰猻huffle subtract(other)第一

26、個(gè)RDD中存在，而不存在與第二個(gè)RDD的元素。需要shuffle。使用場(chǎng)景，機(jī)器學(xué)習(xí)中，移除訓(xùn)練集。,集合運(yùn)算,cartesian(other) 非常耗時(shí)。 使用場(chǎng)景：用戶相似性的時(shí)候,RDD的transformations,基本的RDD transformations： RDD 包含 1, 2, 3, 3,RDD的transformations,兩個(gè)RDD 的transformations： 一個(gè)RDD包含 1, 2, 3，另一個(gè)RDD包含 3, 4, 5,Actions,在RDD上計(jì)算出來(lái)一個(gè)結(jié)果，把結(jié)果返回給driver program或者保存在外部文件系統(tǒng)上，像count() 函數(shù) f

27、irst()。 count() 返回元素的個(gè)數(shù),RDD的actions,Actions,reduce() 最常用的是reduce()，接收一個(gè)函數(shù)，作用在RDD的兩個(gè)類(lèi)型相同的元素上，返回一個(gè)類(lèi)型相同的新元素。 最常用的一個(gè)函數(shù)是加法。 使用reduce()我們可以很簡(jiǎn)單的實(shí)現(xiàn)，RDD中元素的累加，計(jì)數(shù)，和其它類(lèi)型的聚集操作。 例子- reduce() val sum = rdd.reduce(x, y) = x + y),Actions,fold() 與reduce()相似， 類(lèi)型相同 但是，在每個(gè)分區(qū)的初始化調(diào)用的時(shí)候，多了個(gè)“zero value” “zero value”的特點(diǎn)，把它應(yīng)用在你的函數(shù)上，不管多少次，都不改變值(例如：+操作的0，*操作的1)。,Actions,aggregate() 與fold()相似 類(lèi)型可以不同 我們提供想要返回的“zero value”類(lèi)型。 第一個(gè)函數(shù)，RDD中元素累加(每個(gè)節(jié)點(diǎn)只累加本地的結(jié)果)。 第二個(gè)函數(shù)，合并累加器(合并每個(gè)節(jié)點(diǎn)的結(jié)果)。 可以使用ag