RDD簡介

RDD是Spark的核心抽象，全稱為Resillient Distributed Dataset，即彈性分布式數(shù)據(jù)集

RDD在抽象上來說是一種元素集合，包含了數(shù)據(jù)。它是被分區(qū)的，分為多個分區(qū)，每個分區(qū)分布在集群中的不同節(jié)點(diǎn)上(分區(qū)即partition)，從而讓RDD中的數(shù)據(jù)可以被并行操作。(分布式的特性)

RDD通常通過Hadoop上的文件，即HDFS文件，來進(jìn)行創(chuàng)建；有時也可以通過Spark應(yīng)用程序中的集合來創(chuàng)建。

RDD最重要的特性就是，提供了容錯性，可以自動從節(jié)點(diǎn)失敗中恢復(fù)過來。即如果某個節(jié)點(diǎn)上的RDD partition，因?yàn)楣?jié)點(diǎn)故障，導(dǎo)致數(shù)據(jù)丟了，那么RDD會自動通過自己的數(shù)據(jù)來源重新計(jì)算partition。這一切對使用者是透明的。

RDD的數(shù)據(jù)默認(rèn)的情況下是存放在內(nèi)存中的，但是在內(nèi)存資源不足時，Spark會自動將RDD數(shù)據(jù)寫入磁盤。(彈性的特性)

創(chuàng)建RDD的三種方式

在RDD中，通常就代表和包含了Spark應(yīng)用程序的輸入源數(shù)據(jù)。?
當(dāng)我們，在創(chuàng)建了初始的RDD之后，才可以通過Spark Core提供的transformation算子，對該RDD進(jìn)行transformation(轉(zhuǎn)換)操作，來獲取其他的RDD。?
Spark Core為我們提供了三種創(chuàng)建RDD的方式，包括：?
1. 使用程序中的集合創(chuàng)建RDD?
2. 使用本地文件創(chuàng)建RDD?
3. 使用HDFS文件創(chuàng)建RDD

應(yīng)用場景?
1. 使用程序中的集合創(chuàng)建RDD，主要用于進(jìn)行測試，可以在實(shí)際部署到集群運(yùn)行之前，自己使用集合構(gòu)造測試數(shù)據(jù)，來測試后面的spark應(yīng)用的流程?
2. 使用本地文件創(chuàng)建RDD，主要用于的場景為：在本地臨時性地處理一些存儲了大量數(shù)據(jù)的文件?
3. 使用HDFS文件創(chuàng)建RDD，應(yīng)該是最常用的生產(chǎn)環(huán)境處理方式，主要可以針對HDFS上存儲的大數(shù)據(jù)，進(jìn)行離線批處理操作

實(shí)際操作

并行化創(chuàng)建RDD

如果要通過并行化集合來創(chuàng)建RDD，需要針對程序中的集合，調(diào)用SparkContext中的parallelize()方法。Spark會將集合中的數(shù)據(jù)拷貝到集群上去，形成一個分布式的數(shù)據(jù)集合，也就是一個RDD。即：集合中的部分?jǐn)?shù)據(jù)會到一個節(jié)點(diǎn)上，而另一部分?jǐn)?shù)據(jù)會到其它節(jié)點(diǎn)上。然后就可以采用并行的方式來操作這個分布式數(shù)據(jù)集合。

// 并行化創(chuàng)建RDD部分代碼 // 實(shí)現(xiàn)1到5的累加求和 val arr = Array(1,2,3,4,5) val rdd = sc.parallelize(arr) val sum = rdd.reduce(_ + _)

官網(wǎng)的注意點(diǎn)?
通過閱讀Spark的官方文檔，如下圖：

我們可知：?
在調(diào)用parallelize()方法時，有一個重要的參數(shù)可以指定，就是要將集合切分成多少個partition。Spark會為每一個partition運(yùn)行一個task來進(jìn)行處理。Spark官方的建議是，為集群中的每個CPU創(chuàng)建2-4個partition。Spark默認(rèn)會根據(jù)集群的情況來設(shè)置partition的數(shù)量。但是也可以在調(diào)用parallelize()方法時，傳入第二個參數(shù)，來設(shè)置RDD的partition數(shù)量。比如，parallelize(arr, 10)

使用textFile方法，通過本地文件或HDFS創(chuàng)建RDD

Spark是支持使用任何Hadoop支持的存儲系統(tǒng)上的文件創(chuàng)建RDD的，比如說HDFS、Cassandra、HBase以及本地文件。通過調(diào)用SparkContext的textFile()方法，可以針對本地文件或HDFS文件創(chuàng)建RDD。Spark是支持使用任何Hadoop支持的存儲系統(tǒng)上的文件創(chuàng)建RDD的，比如說HDFS、Cassandra、HBase以及本地文件。通過調(diào)用SparkContext的textFile()方法，可以針對本地文件或HDFS文件創(chuàng)建RDD。

// 實(shí)現(xiàn)文件字?jǐn)?shù)統(tǒng)計(jì) // textFile()方法中，輸入本地文件路徑或是HDFS路徑 // HDFS:hdfs://spark1:9000/data.txt // local:/home/hadoop/data.txt val rdd = sc.textFile(“/home/hadoop/data.txt”) val wordCount = rdd.map(line => line.length).reduce(_ + _)

通過本地文件或HDFS創(chuàng)建RDD的幾個注意點(diǎn)?
1. 如果是針對本地文件的話：?
* 如果是在Windows上進(jìn)行本地測試，windows上有一份文件即可；?
* 如果是在Spark集群上針對Linux本地文件，那么需要將文件拷貝到所有worker節(jié)點(diǎn)上（就是在spark-submit上使用—master指定了master節(jié)點(diǎn)，使用standlone模式進(jìn)行運(yùn)行，而textFile()方法內(nèi)仍然使用的是Linux本地文件，在這種情況下，是需要將文件拷貝到所有worker節(jié)點(diǎn)上的）；?
2. Spark的textFile()方法支持針對目錄、壓縮文件以及通配符進(jìn)行RDD創(chuàng)建?
3. Spark默認(rèn)會為hdfs文件的每一個block創(chuàng)建一個partition，但是也可以通過textFile()的第二個參數(shù)手動設(shè)置分區(qū)數(shù)量，只能比block數(shù)量多，不能比block數(shù)量少

Spark支持的其余方法，創(chuàng)建RDD

同樣，通過閱讀Spark的官方文檔，可以知道除了通過使用textFile()方法創(chuàng)建RDD之外，還有幾個其余的方法適用于其它的應(yīng)用場景，如下圖：

SparkContext的textFile()除了可以針對上述幾種普通的文件創(chuàng)建RDD之外，還有一些特例的方法來創(chuàng)建RDD：

SparkContext的wholeTextFiles()方法，可以針對一個目錄中的大量小文件，返回由（fileName,fileContent）組成的pair，即pairRDD，而不是普通的RDD。該方法返回的是文件名字和文件中的具體內(nèi)容；而普通的textFile()方法返回的RDD中，每個元素就是文本中一行文本。

SparkContext的sequenceFileK,V方法，可以針對SequenceFile創(chuàng)建RDD，K和V泛型類型就是SequenceFile的key和value的類型。K和V要求必須是Hadoop的序列化機(jī)制，比如IntWritable、Text等。

SparkContext的hadoopRDD()方法，對于Hadoop的自定義輸入類型，可以創(chuàng)建RDD。該方法接收J(rèn)obConf、InputFormatClass、Key和Value的Class。

SparkContext的objectFile()方法，可以針對之前調(diào)用的RDD的saveAsObjectFile()創(chuàng)建的對象序列化的文件，反序列化文件中的數(shù)據(jù)，并創(chuàng)建一個RDD。

總結(jié)

以上是生活随笔為你收集整理的RDD的几种创建方式的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯，歡迎將生活随笔推薦給好友。