05 RDD

大家好，我是一條~

5小時推開Spark的大門，最后一小時，聊聊提了這么久的RDD。

話不多說，開干！

什么是RDD

Spark為了能夠進行高并發和高吞吐的數據處理，封裝了三大數據結構，用于處理不同的應用場景。分別是：

RDD : 彈性分布式數據集

累加器：分布式共享只寫變量

廣播變量：分布式共享只讀變量

當前的很多框架對迭代式算法場景與交互性數據挖掘場景的處理性能非常差, 這個是RDD的提出的動機。

接下來我們重點看看RDD是如何在數據處理中使用的。

它代表一個不可變、只讀的，被分區的數據集。操作 RDD 就像操作本地集合一樣，有很多的方法可以調用，使用方便，而無需關心底層的調度細節。

五大特性

RDD總共有五個特征，三個基本特征，兩個可選特征。

分區（partition）：有一個數據分片列表，可以將數據進行劃分，切分后的數據能夠進行并行計算，是數據集的原子組成部分。

函數（compute）：對于每一個分片都會有一個函數去迭代/計算執行它。

依賴（dependency）：每一個RDD對父RDD有依賴關系，源RDD沒有依賴，通過依賴關系建立來記錄它們之間的關系。

優先位置（可選）：每一個分片會優先計算位置（prefered location）。即要執行任務在哪幾臺機器上好一點（數據本地性）。

分區策略（可選）：對于key-value的RDD可以告訴它們如何進行分片。可以通過repartition函數進行指定。

執行原理

從計算的角度來講，數據處理過程中需要計算資源（內存 & CPU）和計算模型（邏輯），執行時，需要將計算資源和計算模型進行協調和整合。

Spark 框架在執行時，先申請資源，然后將應用程序的數據處理邏輯分解成一個一個的計算任務。然后將任務發到已經分配資源的計算節點上, 按照指定的計算模型進行數據計算。最后得到計算結果。

創建RDD

在 Spark 中創建 RDD 的創建方式可以分為四種。

打開IDEA，創建一個Scala Class。

1.從內存中創建RDD

Spark 主要提供了兩個方法：parallelize 和 makeRDD

import org.apache.spark.{SparkConf, SparkContext} object Rdd { def main(args: Array[String]): Unit = { val sparkConf = new SparkConf().setMaster("local[*]").setAppName("spark") val sparkContext = new SparkContext(sparkConf) val rdd1 = sparkContext.parallelize( List(1,2,3,4) ) val rdd2 = sparkContext.makeRDD( List(1,2,3,4) ) rdd1.collect().foreach(println) rdd2.collect().foreach(println) sparkContext.stop() } }

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

輸出結果

從底層代碼實現來講，makeRDD方法其實就是parallelize方法。

def makeRDD[T: ClassTag]( seq: Seq[T], numSlices: Int = defaultParallelism): RDD[T] = withScope { parallelize(seq, numSlices) }

1

2

3

4

5

6

7

8

9

2.從外部存儲（文件）創建 RDD

由外部存儲系統的數據集創建 RDD 包括：本地的文件系統，所有 Hadoop 支持的數據集，比如 HDFS、HBase 等。

和第二節提到的讀取文件，統計有多少行是一樣的。如果Windows系統執行如下代碼出現問題，可回顧一下第二節去Spark Shell創建。

import org.apache.spark.rdd.RDD import org.apache.spark.{SparkConf, SparkContext} object Rdd { def main(args: Array[String]): Unit = { val sparkConf = new SparkConf().setMaster("local[*]").setAppName("spark") val sparkContext = new SparkContext(sparkConf) val fileRDD = sparkContext.textFile("src/main/java/test.txt") fileRDD.collect().foreach(println) sparkContext.stop() } }

1

2

3

4

5

6

7

8

9

10

11

12

輸出結果

3.從其他RDD創建

主要是通過一個 RDD 運算完后，再產生新的 RDD。

4.直接創建RDD

使用 new 的方式直接構造 RDD，一般由 Spark 框架自身使用。

最后

恭喜堅持到這里的各位同學，通過5天約5個小時的學習，同學們對Spark有個簡單的了解，還完成了大數據入門經典案例——WordCount。

但是，想要學好Spark仍然任重而道遠，送給同學們我本人很喜歡的一句話：

道阻且長，行則將至。

流水不爭先，爭的是川流不息。

感謝各位5天的支持，在此謝過！最后，祝同學們新年快樂！

spark

版權聲明：本文內容由網絡用戶投稿，版權歸原作者所有，本站不擁有其著作權，亦不承擔相應法律責任。如果您發現本站中有涉嫌抄襲或描述失實的內容，請聯系我們jiasou666@gmail.com 處理，核實后本網站將在24小時內刪除侵權內容。

版權聲明：本文內容由網絡用戶投稿，版權歸原作者所有，本站不擁有其著作權，亦不承擔相應法律責任。如果您發現本站中有涉嫌抄襲或描述失實的內容，請聯系我們jiasou666@gmail.com 處理，核實后本網站將在24小時內刪除侵權內容。