RDD簡介

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

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

RDD通常通過Hadoop上的文件，即HDFS文件，來進行創建；有時也可以通過Spark應用程序中的集合來創建。

RDD最重要的特性就是，提供了容錯性，可以自動從節點失敗中恢復過來。即如果某個節點上的RDD partition，因為節點故障，導致數據丟了，那么RDD會自動通過自己的數據來源重新計算partition。這一切對使用者是透明的。

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

創建RDD的三種方式

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

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

實際操作

并行化創建RDD

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

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

官網的注意點?
通過閱讀Spark的官方文檔，如下圖：

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

使用textFile方法，通過本地文件或HDFS創建RDD

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

// 實現文件字數統計 // 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創建RDD的幾個注意點?
1. 如果是針對本地文件的話：?
* 如果是在Windows上進行本地測試，windows上有一份文件即可；?
* 如果是在Spark集群上針對Linux本地文件，那么需要將文件拷貝到所有worker節點上（就是在spark-submit上使用—master指定了master節點，使用standlone模式進行運行，而textFile()方法內仍然使用的是Linux本地文件，在這種情況下，是需要將文件拷貝到所有worker節點上的）；?
2. Spark的textFile()方法支持針對目錄、壓縮文件以及通配符進行RDD創建?
3. Spark默認會為hdfs文件的每一個block創建一個partition，但是也可以通過textFile()的第二個參數手動設置分區數量，只能比block數量多，不能比block數量少

Spark支持的其余方法，創建RDD

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

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

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

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

SparkContext的hadoopRDD()方法，對于Hadoop的自定義輸入類型，可以創建RDD。該方法接收JobConf、InputFormatClass、Key和Value的Class。

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

總結

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

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。