Spark大數(shù)據(jù)分析 課件 第4章 彈性分布式數(shù)據(jù)集

RDD概述目錄/Contents01

RDD概述RDD概述01RDD概述RDD（ResilientDistributedDataset）叫做分布式數(shù)據(jù)集，是Spark中最基本的數(shù)據(jù)抽象，它代表一個(gè)不可變、可分區(qū)、里面的元素可并行計(jì)算的集合。在Spark中，對(duì)數(shù)據(jù)的所有操作不外乎創(chuàng)建RDD、轉(zhuǎn)化已有RDD以及調(diào)用RDD操作進(jìn)行求值。每個(gè)RDD都被分為多個(gè)分區(qū)，這些分區(qū)運(yùn)行在集群中的不同節(jié)點(diǎn)上。RDD可以包含Python、Java、Scala中任意類型的對(duì)象，甚至可以包含用戶自定義的對(duì)象。RDD具有數(shù)據(jù)流模型的特點(diǎn)：自動(dòng)容錯(cuò)、位置感知性調(diào)度和可伸縮性。RDD允許用戶在執(zhí)行多個(gè)查詢時(shí)顯式地將工作集緩存在內(nèi)存中，后續(xù)的查詢能夠重用工作集，這極大地提升了查詢速度。RDD支持兩種操作:transformation操作和action操作。RDD的轉(zhuǎn)化操作是返回一個(gè)新的RDD的操作，比如map()和filter()，而action操作則是向驅(qū)動(dòng)器程序返回結(jié)果或把結(jié)果寫入外部系統(tǒng)的操作。比如count()和first()。RDD概述RDD具有如下五個(gè)特性：1、一組分片（Partition），即數(shù)據(jù)集的基本組成單位。對(duì)于RDD來說，每個(gè)分片都會(huì)被一個(gè)計(jì)算任務(wù)處理，并決定并行計(jì)算的粒度。2、一個(gè)計(jì)算每個(gè)分區(qū)的函數(shù)。Spark中RDD的計(jì)算是以分片為單位的，每個(gè)RDD都會(huì)實(shí)現(xiàn)compute函數(shù)以達(dá)到這個(gè)目的。3、RDD之間的依賴關(guān)系。RDD的每次轉(zhuǎn)換都會(huì)生成一個(gè)新的RDD，所以RDD之間就會(huì)形成類似于流水線一樣的前后依賴關(guān)系。在部分分區(qū)數(shù)據(jù)丟失時(shí)，Spark可以通過這個(gè)依賴關(guān)系重新計(jì)算丟失的分區(qū)數(shù)據(jù)，而不是對(duì)RDD的所有分區(qū)進(jìn)行重新計(jì)算。RDD概述4、一個(gè)Partitioner，即RDD的分片函數(shù)（分區(qū)器）。當(dāng)前Spark中實(shí)現(xiàn)了兩種類型的分片函數(shù)，一個(gè)是基于哈希的HashPartitioner，另外一個(gè)是基于范圍的RangePartitioner。只有對(duì)于于key-value的RDD，才會(huì)有Partitioner，非key-value的RDD的Parititioner的值是None。Partitioner函數(shù)不但決定了RDD本身的分片數(shù)量，也決定了parentRDDShuffle輸出時(shí)的分片數(shù)量。5、一個(gè)列表，存儲(chǔ)存取每個(gè)Partition的優(yōu)先位置（preferredlocation）。對(duì)于一個(gè)HDFS文件來說，這個(gè)列表保存的就是每個(gè)Partition所在的塊的位置。按照“移動(dòng)數(shù)據(jù)不如移動(dòng)計(jì)算”的理念，Spark在進(jìn)行任務(wù)調(diào)度的時(shí)候，會(huì)盡可能地將計(jì)算任務(wù)分配到其所要處理數(shù)據(jù)塊的存儲(chǔ)位置。感謝大家的聆聽RDD創(chuàng)建方式目錄/Contents01通過讀取文件生成RDD02通過并行化方式創(chuàng)建RDD通過讀取文件生成RDD01通過讀取文件生成RDDSpark是支持使用任何Hadoop支持的存儲(chǔ)系統(tǒng)上的文件創(chuàng)建RDD的，比如說HDFS、Cassandra、HBase以及本地文件。通過調(diào)用SparkContext的textFile()方法，可以針對(duì)本地文件或HDFS文件創(chuàng)建RDD。1．通過加載本地文件數(shù)據(jù)創(chuàng)建RDD2．通過加載HDFS文件數(shù)據(jù)創(chuàng)建RDD通過并行化方式創(chuàng)建RDD02通過并行化方式創(chuàng)建RDD如果要通過并行化集合來創(chuàng)建RDD，需要針對(duì)程序中已經(jīng)存在的集合、數(shù)組，調(diào)用SparkContext中的parallelize()方法。Spark會(huì)將集合中的數(shù)據(jù)拷貝到集群上去，形成一個(gè)分布式的數(shù)據(jù)集合，也就是一個(gè)RDD。即：集合中的部分?jǐn)?shù)據(jù)會(huì)到一個(gè)節(jié)點(diǎn)上，而另一部分?jǐn)?shù)據(jù)會(huì)到其它節(jié)點(diǎn)上。然后就可以采用并行的方式來操作這個(gè)分布式數(shù)據(jù)集合。感謝大家的聆聽RDD類型操作目錄/Contents01

轉(zhuǎn)換算子02行動(dòng)算子轉(zhuǎn)換算子01轉(zhuǎn)換算子Transformation：轉(zhuǎn)換算子，這類轉(zhuǎn)換并不觸發(fā)提交作業(yè)，完成作業(yè)中間過程處理。下面是一些常用的轉(zhuǎn)換算子操作的API。操作介紹map(func)將RDD中的每個(gè)元素傳入自定義函數(shù)，獲取一個(gè)新的元素，然后用新的元素組成新的RDDfilter(func)對(duì)RDD中每個(gè)元素進(jìn)行判斷，如果返回true則保留，返回false則剔除flatMap(func)與map類似，但是對(duì)每個(gè)元素都可以返回一個(gè)或多個(gè)新元素groupByKey(func)根據(jù)key進(jìn)行分組，每個(gè)key對(duì)應(yīng)一個(gè)Iterable<value>reduceByKey(func)對(duì)每個(gè)key對(duì)應(yīng)value進(jìn)行reduce操作轉(zhuǎn)換算子1.map(func)map(func)操作是對(duì)RDD中的每個(gè)元素都執(zhí)行一個(gè)指定的函數(shù)來產(chǎn)生一個(gè)新的RDD。2.filter(func)filter(func)操作會(huì)篩選出滿足條件（即func）的元素，返回一個(gè)新的數(shù)據(jù)集。3.flatMap(func)類似于map，但是每一個(gè)輸入元素，會(huì)被映射為0到多個(gè)輸出元素(因此，func函數(shù)的返回值是一個(gè)Seq，而不是單一元素)。在一個(gè)由(K,V)對(duì)組成的數(shù)據(jù)集上調(diào)用，返回一個(gè)(K，Seq[V])對(duì)的數(shù)據(jù)集。4.groupByKey(func)5.reduceByKey(func)顧名思義，reduceByKey就是對(duì)元素為KV對(duì)的RDD中Key相同的元素的Value進(jìn)行reduce，因此，Key相同的多個(gè)元素的值被reduce為一個(gè)值，然后與原RDD中的Key組成一個(gè)新的KV對(duì)。行動(dòng)算子02行動(dòng)算子Action：行動(dòng)算子，這類算子會(huì)觸發(fā)SparkContext提交Job作業(yè)。下面是一些常用的行動(dòng)算子操作的API。操作介紹

reduce(func)通過函數(shù)func聚集數(shù)據(jù)集中的所有元素。Func函數(shù)接受2個(gè)參數(shù)，返回一個(gè)值。這個(gè)函數(shù)必須是關(guān)聯(lián)性的，確保可以被正確的并發(fā)執(zhí)行collect()在Driver的程序中，以數(shù)組的形式，返回?cái)?shù)據(jù)集的所有元素。count()返回?cái)?shù)據(jù)集的元素個(gè)數(shù)take(n)返回一個(gè)數(shù)組，由數(shù)據(jù)集的前n個(gè)元素組成。first()返回?cái)?shù)據(jù)集的第一個(gè)元素（類似于take（1）)foreach(func)在數(shù)據(jù)集的每一個(gè)元素上，運(yùn)行函數(shù)func。

saveAsTextFile(path)將數(shù)據(jù)集的元素，以textfile的形式，保存到本地文件系統(tǒng)，hdfs或者任何其它hadoop支持的文件系統(tǒng)。Spark將會(huì)調(diào)用每個(gè)元素的toString方法，并將它轉(zhuǎn)換為文件中的一行文本行動(dòng)算子1.reduce(func)reduce將RDD中元素兩兩傳遞給輸入函數(shù)，同時(shí)產(chǎn)生一個(gè)新的值，新產(chǎn)生的值與RDD中下一個(gè)元素再被傳遞給輸入函數(shù)直到最后只有一個(gè)值為止。2.collect()在Driver的程序中，以數(shù)組的形式，返回?cái)?shù)據(jù)集的所有元素。這通常會(huì)在使用filter或者其它操作后，返回一個(gè)足夠小的數(shù)據(jù)子集再使用，直接將整個(gè)RDD集Collect返回，很可能會(huì)讓Driver程序OOM。3.count()count返回整個(gè)RDD的元素個(gè)數(shù)。我們可以定義一個(gè)RDD，使用count()來統(tǒng)計(jì)RDD的元素個(gè)數(shù)。

take和collect操作類似，只是collect操作獲取的所有數(shù)據(jù)，而take操作是獲取前n個(gè)元素。4.take(n)5.first()first()的作用是返回?cái)?shù)據(jù)集的第一個(gè)元素.我們可以定義一個(gè)RDD,使用first()來獲取RDD中的第一個(gè)元素。6.foreach(func)foreach對(duì)RDD中的每個(gè)元素都應(yīng)用func函數(shù)操作，不返回RDD和Array，而是返回Uint。感謝大家的聆聽RDD之間的依賴關(guān)系目錄/Contents01

RDD之間的依賴關(guān)系RDD之間的依賴關(guān)系01RDD之間的依賴關(guān)系RDD和它依賴的父RDD的關(guān)系有兩種不同的類型，即窄依賴（narrowdependency）和寬依賴（widedependency）。RDD之間的依賴關(guān)系窄依賴指的是每一個(gè)父RDD的Partition最多被子RDD的一個(gè)Partition使用?？偨Y(jié)：窄依賴我們形象的比喻為獨(dú)生子女。1、窄依賴RDD之間的依賴關(guān)系寬依賴指的是多個(gè)子RDD的Partition會(huì)依賴同一個(gè)父RDD的Partition?？偨Y(jié)：寬依賴我們形象的比喻為超生。2、寬依賴RDD之間的依賴關(guān)系RDD只支持粗粒度轉(zhuǎn)換，即只記錄單個(gè)塊上執(zhí)行的單個(gè)操作。將創(chuàng)建RDD的一系列Lineage（即血統(tǒng)）記錄下來，以便恢復(fù)丟失的分區(qū)。RDD的Lineage會(huì)記錄RDD的元數(shù)據(jù)信息和轉(zhuǎn)換行為，當(dāng)該RDD的部分分區(qū)數(shù)據(jù)丟失時(shí)，它可以根據(jù)這些信息來重新運(yùn)算和恢復(fù)丟失的數(shù)據(jù)分區(qū)。3、Lineage（血統(tǒng)）感謝大家的聆聽RDD機(jī)制目錄/Contents01持久化機(jī)制01容錯(cuò)機(jī)制持久化機(jī)制01持久化機(jī)制多次對(duì)某個(gè)RDD進(jìn)行transformation或者action，如果沒有做RDD持久化，那么每次都要重新計(jì)算一個(gè)RDD，會(huì)消耗大量時(shí)間，降低Spark性能。Spark非常重要的一個(gè)功能特性就是可以將RDD持久化在內(nèi)存中。當(dāng)對(duì)RDD執(zhí)行持久化操作時(shí)，每個(gè)節(jié)點(diǎn)都會(huì)將自己操作的RDD的partition持久化到內(nèi)存中，并且在之后對(duì)該RDD的反復(fù)使用中，直接使用內(nèi)存緩存的partition。這樣的話，對(duì)于針對(duì)一個(gè)RDD反復(fù)執(zhí)行多個(gè)操作的場景，就只要對(duì)RDD計(jì)算一次即可，后面直接使用該RDD，而不需要反復(fù)計(jì)算多次該RDD。持久化機(jī)制存儲(chǔ)級(jí)別說明

MEMORY_ONLY將RDD以反序列化Java對(duì)象的形式存儲(chǔ)在JVM中。如果內(nèi)存空間不夠，部分分區(qū)將不再緩存，在每次需要用到這些數(shù)據(jù)時(shí)重新進(jìn)行計(jì)算。這是默認(rèn)的存儲(chǔ)級(jí)別。

MEMORY_AND_DISK將RDD以反序列化Java對(duì)象的形式存儲(chǔ)在JVM中。如果內(nèi)存空間不夠，將未緩存的數(shù)據(jù)分區(qū)存儲(chǔ)到磁盤，在需要使用這些分區(qū)時(shí)從磁盤讀取。

MEMORY_ONLY_SER將RDD以序列化的Java對(duì)象的形式進(jìn)行存儲(chǔ)（每個(gè)分區(qū)為一個(gè)字節(jié)數(shù)組）。這種方式比反序列化的Java對(duì)象節(jié)省空間，但是在讀取時(shí)會(huì)增加CPU的計(jì)算負(fù)擔(dān)。

MEMORY_AND_DISK_SER類似于MEMORY_ONLY_SER，但是溢出的分區(qū)會(huì)存儲(chǔ)到磁盤，而不是在用到它們時(shí)重新計(jì)算。DISK_ONLY只在磁盤上緩存RDD。

MEMORY_ONLY_2，MEMORY_AND_DISK_2與上面的級(jí)別功能相同，只不過每個(gè)分區(qū)在集群中兩個(gè)節(jié)點(diǎn)上建立副本。需要加上后綴_2,代表的是將每個(gè)持久化的數(shù)據(jù)都復(fù)制一份副本，并將副本保存到其他節(jié)點(diǎn)上。

OFF_HEAP類似于MEMORY_ONLY_SER，但是將數(shù)據(jù)存儲(chǔ)在off-heapmemory（堆外內(nèi)存），這需要啟動(dòng)off-heap內(nèi)存。持久化機(jī)制1.如何選擇存儲(chǔ)級(jí)別Spark的存儲(chǔ)級(jí)別的選擇，核心問題是在內(nèi)存使用率和CPU效率之間進(jìn)行權(quán)衡。2.cache()和persist()的區(qū)別cache()和persist()的區(qū)別在于，cache()是persist()的一種簡化方式，cache()的底層就是調(diào)用的persist()的無參版本，同時(shí)就是調(diào)用persist(MEMORY_ONLY)，將數(shù)據(jù)持久化到內(nèi)存中。容錯(cuò)機(jī)制02容錯(cuò)機(jī)制Spark的計(jì)算本質(zhì)就是對(duì)RDD做各種轉(zhuǎn)換，因?yàn)镽DD是一個(gè)不可變只讀的集合，因此每次的轉(zhuǎn)換都需要上一次的RDD作為本次轉(zhuǎn)換的輸入，因此RDD的lineage描述的是RDD間的相互依賴關(guān)系。為了保證RDD中數(shù)據(jù)的健壯性，RDD數(shù)據(jù)集通過所謂血統(tǒng)關(guān)系（lineage）記住了他是如何其他RDD中演變過來的。Spark將RDD之間的關(guān)系規(guī)類為寬依賴和窄依賴。Spark會(huì)根據(jù)Lineage存儲(chǔ)的RDD的依賴關(guān)系對(duì)RDD計(jì)算做故障容錯(cuò)，目前Spark的容錯(cuò)策略主要是根據(jù)RDD依賴關(guān)系重新計(jì)算、對(duì)RDD做cache、對(duì)RDD做checkpoint手段完成RDD計(jì)算的故障容錯(cuò)。容錯(cuò)機(jī)制SparkRDD實(shí)現(xiàn)基于Lineage的容錯(cuò)機(jī)制，基于RDD的各項(xiàng)transformation構(gòu)成了computechain，在部分計(jì)算結(jié)果丟失的時(shí)候可以根據(jù)Lineage重新恢復(fù)計(jì)算。

（1）在窄依賴中，在子RDD的分區(qū)丟失，要重算父RDD分區(qū)時(shí)，父RDD相應(yīng)分區(qū)的所有數(shù)據(jù)都是子RDD分區(qū)的數(shù)據(jù)，并不存在冗余計(jì)算。（2）在寬依賴情況下，丟失一個(gè)子RDD分區(qū)，重算的每個(gè)父RDD的每個(gè)分區(qū)的所有數(shù)據(jù)并不是都給丟失的子RDD分區(qū)用的，會(huì)有一部分?jǐn)?shù)據(jù)相當(dāng)于對(duì)應(yīng)的是未丟失的子RDD分區(qū)中需要的數(shù)據(jù)，這樣就會(huì)產(chǎn)生冗余計(jì)算開銷和巨大的性能浪費(fèi)。感謝大家的聆聽SparkSQL簡介目錄/Contents01什么是SparkSQL02SparkSQL架構(gòu)什么是SparkSQL01什么是SparkSQLSparkSQL架構(gòu)與Hive架構(gòu)相比除了把底層的MapReduce執(zhí)行引擎更改為Spark還