1. 程式人生 > >Spark介紹(二)RDD

Spark介紹(二)RDD

一、RDD介紹

彈性分散式資料集,RDDSpark最核心的東西,它表示已被分割槽,不可變的並能夠被並行操作的資料集合,不同的資料集格式對應不同的RDD實現。

RDD特點:

1.來源:一種是從持久儲存獲取資料(並行化集合或Hadoop資料集),另一種是從其他RDD生成

2.只讀:狀態不可變,不能修改

3.分割槽:支援元素根據 Key 來分割槽 ( Partitioning ) ,儲存到多個結點上,還原時只會重新計算丟失分割槽的資料,而不會影響整個系統

4.路徑:在 RDD 中叫世族或血統 ( lineage ) 

,即 RDD 有充足的資訊關於它是如何從其他 RDD 產生而來的,DAG反應RDD之間的依賴關係

5.持久化:可以控制儲存級別(記憶體、磁碟等)來進行持久化(cache,persist

6.操作:豐富的動作 ( Action ) ,如CountReduceCollectSave 

共享變數:廣播變數sc.broadcast(v)和累加器sc.accumulator(v)

二、RDD轉換與操作

對於RDD可以有兩種計算方式:轉換(返回值還是一個RDD)與操作(返回值不是一個RDD

1.轉換(Transformations) (

如:map, filter, groupBy, join)Transformations操作是Lazy的,也就是說從一個RDD轉換生成另一個RDD的操作不是馬上執行,Spark在遇到Transformations操作時只會記錄需要這樣的操作,並不會去執行,需要等到有Actions操作的時候才會真正啟動計算過程進行計算。

2.操作(Actions) (如:count, collect, save)Actions操作會返回結果或把RDD資料寫到儲存系統中。Actions是觸發Spark啟動計算的動因。

                          

三、RDD操作例子

如何建立RDD

RDD可以從普通陣列創建出來,也可以從檔案系統或者HDFS中的檔案創建出來。

舉例:從普通陣列建立RDD,裡面包含了199個數字,它們分別在3個分割槽中。

scala> val a = sc.parallelize(1 to 9, 3)

a: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[1] at parallelize at <console>:12

舉例:讀取檔案README.md來建立RDD,檔案中的每一行就是RDD中的一個元素

scala> val b = sc.textFile("README.md")

b: org.apache.spark.rdd.RDD[String] = MappedRDD[3] at textFile at <console>:12

雖然還有別的方式可以建立RDD,但在本文中我們主要使用上述兩種方式來建立RDD以說明RDDAPI

map

map是對RDD中的每個元素都執行一個指定的函式來產生一個新的RDD任何原RDD中的元素在新RDD中都有且只有一個元素與之對應。

舉例:

scala> val a = sc.parallelize(1 to 9, 3)
scala> val b = a.map(x => x*2)
scala> a.collect
res10: Array[Int] = Array(1, 2, 3, 4, 5, 6, 7, 8, 9)
scala> b.collect
res11: Array[Int] = Array(2, 4, 6, 8, 10, 12, 14, 16, 18)

上述例子中把原RDD中每個元素都乘以2來產生一個新的RDD

mapPartitions

mapPartitionsmap的一個變種。map的輸入函式是應用於RDD中每個元素,而mapPartitions的輸入函式是應用於每個分割槽,也就是把每個分割槽中的內容作為整體來處理的。
它的函式定義為:

def mapPartitions[U: ClassTag](f: Iterator[T] => Iterator[U], preservesPartitioning: Boolean = false): RDD[U]

f即為輸入函式,它處理每個分割槽裡面的內容。每個分割槽中的內容將以Iterator[T]傳遞給輸入函式ff的輸出

結果是Iterator[U]。最終的RDD由所有分割槽經過輸入函式處理後的結果合併起來的。

舉例:

scala> val a = sc.parallelize(1 to 9, 3)
scala> def myfunc[T](iter: Iterator[T]) : Iterator[(T, T)] = {
    var res = List[(T, T)]() 
    var pre = iter.next while (iter.hasNext) {
        val cur = iter.next; 
        res .::= (pre, cur) pre = cur;
    } 
    res.iterator
}
scala> a.mapPartitions(myfunc).collect
res0: Array[(Int, Int)] = Array((2,3), (1,2), (5,6), (4,5), (8,9), (7,8))

上述例子中的函式myfunc是把分割槽中一個元素和它的下一個元素組成一個Tuple。因為分割槽中最後一個元素沒有下一個元素了,所以(3,4)(6,7)不在結果中。
mapPartitions還有些變種,比如mapPartitionsWithContext,它能把處理過程中的一些狀態資訊傳遞給使用者指定的輸入函式。還有mapPartitionsWithIndex,它能把分割槽的index傳遞給使用者指定的輸入函式。

mapValues

mapValues顧名思義就是輸入函式應用於RDDKev-ValueValue,原RDD中的Key保持不變,與新的Value一起組成新的RDD中的元素。因此,該函式只適用於元素為KV對的RDD

舉例:

scala> val a = sc.parallelize(List("dog", "tiger", "lion", "cat", "panther", " eagle"), 2)

scala> val b = a.map(x => (x.length, x))

scala> b.mapValues("x" + _ + "x").collect

res5: Array[(Int, String)] = Array((3,xdogx), (5,xtigerx), (4,xlionx),(3,xcatx), (7,xpantherx), (5,xeaglex))

mapWith

mapWithmap的另外一個變種,map只需要一個輸入函式,而mapWith有兩個輸入函式。它的定義如下:

def mapWith[A: ClassTag, U: ](constructA: Int => A, preservesPartitioning: Boolean = false)(f: (T, A) => U): RDD[U]

  • 第一個函式constructA是把RDD的partition index(index從0開始)作為輸入,輸出為新型別A;
  • 第二個函式f是把二元組(T, A)作為輸入(其中T為原RDD中的元素,A為第一個函式的輸出),輸出型別為U。

舉例:把partition index 乘以10,然後加上2作為新的RDD的元素。

val x = sc.parallelize(List(1,2,3,4,5,6,7,8,9,10), 3)

x.mapWith(a => a * 10)((a, b) => (b + 2)).collect

res4: Array[Int] = Array(2, 2, 2, 12, 12, 12, 22, 22, 22, 22)

flatMap

map類似,區別是原RDD中的元素經map處理後只能生成一個元素,而原RDD中的元素經flatmap處理後可生成多個元素來構建新RDD 
舉例:對原RDD中的每個元素x產生y個元素(從1yy為元素x的值)

scala> val a = sc.parallelize(1 to 4, 2)

scala> val b = a.flatMap(x => 1 to x)

scala> b.collect

res12: Array[Int] = Array(1, 1, 2, 1, 2, 3, 1, 2, 3, 4)

flatMapWith

flatMapWithmapWith很類似,都是接收兩個函式,一個函式把partitionIndex作為輸入,輸出是一個新型別A;另外一個函式是以二元組(T,A)作為輸入,輸出為一個序列,這些序列裡面的元素組成了新的RDD。它的定義如下:

def flatMapWith[A: ClassTag, U: ClassTag](constructA: Int => A, preservesPartitioning: Boolean = false)(f: (T, A) => Seq[U]): RDD[U]

舉例:

scala> val a = sc.parallelize(List(1,2,3,4,5,6,7,8,9), 3)

scala> a.flatMapWith(x => x, true)((x, y) => List(y, x)).collect

res58: Array[Int] = Array(0, 1, 0, 2, 0, 3, 1, 4, 1, 5, 1, 6, 2, 7, 2, 8, 2, 9)

flatMapValues

flatMapValues類似於mapValues,不同的在於flatMapValues應用於元素為KV對的RDDValue。每個一元素的Value被輸入函式對映為一系列的值,然後這些值再與原RDD中的Key組成一系列新的KV對。

舉例

scala> val a = sc.parallelize(List((1,2),(3,4),(3,6)))

scala> val b = a.flatMapValues(x=>x.to(5))

scala> b.collect

res3: Array[(Int, Int)] = Array((1,2), (1,3), (1,4), (1,5), (3,4), (3,5))

上述例子中原RDD中每個元素的值被轉換為一個序列(從其當前值到5),比如第一個KV(1,2), 其值2被轉換為2345。然後其再與原KV對中Key組成一系列新的KV(1,2),(1,3),(1,4),(1,5)

reduce

reduceRDD中元素兩兩傳遞給輸入函式,同時產生一個新的值,新產生的值與RDD中下一個元素再被傳遞給輸入函式直到最後只有一個值為止。

舉例

scala> val c = sc.parallelize(1 to 10)

scala> c.reduce((x, y) => x + y)

res4: Int = 55

上述例子對RDD中的元素求和。

reduceByKey

顧名思義,reduceByKey就是對元素為KV對的RDDKey相同的元素的Value進行reduce,因此,Key相同的多個元素的值被reduce為一個值,然後與原RDD中的Key組成一個新的KV對。

舉例:

scala> val a = sc.parallelize(List((1,2),(3,4),(3,6)))

scala> a.reduceByKey((x,y) => x + y).collect

res7: Array[(Int, Int)] = Array((1,2), (3,10))

上述例子中,對Key相同的元素的值求和,因此Key3的兩個元素被轉為了(3,10)

更多的API詳情可以檢視Zhenghe的介紹 http://homepage.cs.latrobe.edu.au/zhe/ZhenHeSparkRDDAPIExamples.html