碼上歡樂
相關內容    簡體    繁體

Spark函數詳解系列之RDD基本轉換

本文轉載自   查看原文   2016-04-10 01:51   73179    Spark
摘要：
  RDD：彈性分布式數據集，是一種特殊集合 ‚ 支持多種來源 ‚ 有容錯機制 ‚ 可以被緩存 ‚ 支持並行操作，一個RDD代表一個分區里的數據集
  RDD有兩種操作算子：
         Transformation（轉換）：Transformation屬於延遲計算，當一個RDD轉換成另一個RDD時並沒有立即進行轉換，僅僅是記住了數據集的邏輯操作
         Ation（執行）：觸發Spark作業的運行，真正觸發轉換算子的計算
 
本系列主要講解Spark中常用的函數操作：
         1.RDD基本轉換
         2.鍵-值RDD轉換
         3.Action操作篇
本節所講函數
 
基礎轉換操作：
 
1.map(func)：數據集中的每個元素經過用戶自定義的函數轉換形成一個新的RDD，新的RDD叫MappedRDD
（例1） 
object Map {
  def main(args: Array[String]) {
    val conf = new SparkConf().setMaster("local").setAppName("map")
    val sc = new SparkContext(conf)
    val rdd = sc.parallelize(1 to 10)  //創建RDD
    val map = rdd.map(_*2)             //對RDD中的每個元素都乘於2
    map.foreach(x => print(x+" "))
    sc.stop()
  }
}
輸出： 
2 4 6 8 10 12 14 16 18 20
(RDD依賴圖：紅色塊表示一個RDD區，黑色塊表示該分區集合，下同)
 
2.flatMap(func):與map類似，但每個元素輸入項都可以被映射到0個或多個的輸出項，最終將結果”扁平化“后輸出
（例2） 
 //...省略sc
    val rdd = sc.parallelize(1 to 5)
    val fm = rdd.flatMap(x => (1 to x)).collect()
    fm.foreach( x => print(x + " "))
輸出： 
1 1 2 1 2 3 1 2 3 4 1 2 3 4 5
如果是map函數其輸出如下： 
Range(1) Range(1, 2) Range(1, 2, 3) Range(1, 2, 3, 4) Range(1, 2, 3, 4, 5)

 (RDD依賴圖)

 
3.mapPartitions(func):類似與map，map作用於每個分區的每個元素，但mapPartitions作用於每個分區工
func的類型：Iterator[T] => Iterator[U]
假設有N個元素，有M個分區，那么map的函數的將被調用N次,而mapPartitions被調用M次,當在映射的過程中不斷的創建對象時就可以使用mapPartitions比map的效率要高很多，比如當向數據庫寫入數據時，如果使用map就需要為每個元素創建connection對象，但使用mapPartitions的話就需要為每個分區創建connetcion對象
(例3)：輸出有女性的名字： 
object MapPartitions {
//定義函數 
  def partitionsFun(/*index : Int,*/iter : Iterator[(String,String)]) : Iterator[String] = {
    var woman = List[String]()
    while (iter.hasNext){
      val next = iter.next()
      next match {
        case (_,"female") => woman = /*"["+index+"]"+*/next._1 :: woman
        case _ =>
      }
    }
    return  woman.iterator
  }

  def main(args: Array[String]) {
    val conf = new SparkConf().setMaster("local").setAppName("mappartitions")
    val sc = new SparkContext(conf)
    val l = List(("kpop","female"),("zorro","male"),("mobin","male"),("lucy","female"))
    val rdd = sc.parallelize(l,2)
    val mp = rdd.mapPartitions(partitionsFun)
    /*val mp = rdd.mapPartitionsWithIndex(partitionsFun)*/
    mp.collect.foreach(x => (print(x +" ")))   //將分區中的元素轉換成Aarray再輸出
  }
}
輸出： 
kpop lucy
其實這個效果可以用一條語句完成 
val mp = rdd.mapPartitions(x => x.filter(_._2 == "female")).map(x => x._1)　
之所以不那么做是為了演示函數的定義
  (RDD依賴圖)
 
4.mapPartitionsWithIndex(func):與mapPartitions類似，不同的時函數多了個分區索引的參數
func類型：(Int, Iterator[T]) => Iterator[U]
（例4）：將例3橙色的注釋部分去掉即是
輸出：（帶了分區索引） 
[0]kpop [1]lucy
 
5.sample(withReplacement,fraction,seed):以指定的隨機種子隨機抽樣出數量為fraction的數據，withReplacement表示是抽出的數據是否放回，true為有放回的抽樣，false為無放回的抽樣
(例5)：從RDD中隨機且有放回的抽出50%的數據，隨機種子值為3（即可能以1 2 3的其中一個起始值） 
//省略
    val rdd = sc.parallelize(1 to 10)
    val sample1 = rdd.sample(true,0.5,3)
    sample1.collect.foreach(x => print(x + " "))
    sc.stop

 

6.union(ortherDataset):將兩個RDD中的數據集進行合並，最終返回兩個RDD的並集，若RDD中存在相同的元素也不會去重 
//省略sc
   val rdd1 = sc.parallelize(1 to 3)
   val rdd2 = sc.parallelize(3 to 5)
   val unionRDD = rdd1.union(rdd2)
   unionRDD.collect.foreach(x => print(x + " "))
   sc.stop　
輸出： 
1 2 3 3 4 5

　　

7.intersection(otherDataset):返回兩個RDD的交集 
//省略sc
val rdd1 = sc.parallelize(1 to 3)
val rdd2 = sc.parallelize(3 to 5)
val unionRDD = rdd1.intersection(rdd2)
unionRDD.collect.foreach(x => print(x + " "))
sc.stop　
輸出： 
3 4

 

8.distinct([numTasks]):對RDD中的元素進行去重 
//省略sc
val list = List(1,1,2,5,2,9,6,1)
val distinctRDD = sc.parallelize(list)
val unionRDD = distinctRDD.distinct()
unionRDD.collect.foreach(x => print(x + " "))　　
輸出： 
1 6 9 5 2

 

9.cartesian(otherDataset):對兩個RDD中的所有元素進行笛卡爾積操作 
 //省略
val rdd1 = sc.parallelize(1 to 3)
val rdd2 = sc.parallelize(2 to 5)
val cartesianRDD = rdd1.cartesian(rdd2)
cartesianRDD.foreach(x => println(x + " "))　
輸出： 
(1,2)
(1,3)
(1,4)
(1,5)
(2,2)
(2,3)
(2,4)
(2,5)
(3,2)
(3,3)
(3,4)
(3,5)

 (RDD依賴圖)

 

 
10.coalesce(numPartitions，shuffle):對RDD的分區進行重新分區，shuffle默認值為false,當shuffle=false時，不能增加分區數
目,但不會報錯，只是分區個數還是原來的
(例9:） shuffle=false 
//省略  
val rdd = sc.parallelize(1 to 16,4)
val coalesceRDD = rdd.coalesce(3) //當suffle的值為false時，不能增加分區數(即分區數不能從5->7)
println("重新分區后的分區個數:"+coalesceRDD.partitions.size)　
輸出： 
重新分區后的分區個數:3
//分區后的數據集
List(1, 2, 3, 4)
List(5, 6, 7, 8)
List(9, 10, 11, 12, 13, 14, 15, 16)

　

(例9.1:） shuffle=true 
 //...省略
val rdd = sc.parallelize(1 to 16,4)
val coalesceRDD = rdd.coalesce(7,true)
println("重新分區后的分區個數:"+coalesceRDD.partitions.size)
println("RDD依賴關系:"+coalesceRDD.toDebugString)　　
輸出： 
重新分區后的分區個數:5
RDD依賴關系:(5) MapPartitionsRDD[4] at coalesce at Coalesce.scala:14 []
| CoalescedRDD[3] at coalesce at Coalesce.scala:14 []
| ShuffledRDD[2] at coalesce at Coalesce.scala:14 []
+-(4) MapPartitionsRDD[1] at coalesce at Coalesce.scala:14 []
| ParallelCollectionRDD[0] at parallelize at Coalesce.scala:13 []
//分區后的數據集
List(10, 13)
List(1, 5, 11, 14)
List(2, 6, 12, 15)
List(3, 7, 16)
List(4, 8, 9)

 (RDD依賴圖:coalesce(3,flase))

 

 

  (RDD依賴圖:coalesce(3,true))
 
 
11.repartition(numPartition):是函數coalesce(numPartition,true)的實現，效果和例9.1的coalesce(numPartition,true)的一樣
 
 
12.glom():將RDD的每個分區中的類型為T的元素轉換換數組Array[T]
  
//省略
val rdd = sc.parallelize(1 to 16,4)
val glomRDD = rdd.glom() //RDD[Array[T]]
glomRDD.foreach(rdd => println(rdd.getClass.getSimpleName))
sc.stop　
輸出： 
int[] //說明RDD中的元素被轉換成數組Array[Int]
 
 
13.randomSplit(weight:Array[Double],seed):根據weight權重值將一個RDD划分成多個RDD,權重越高划分得到的元素較多的幾率就越大 
//省略sc
val rdd = sc.parallelize(1 to 10)
val randomSplitRDD = rdd.randomSplit(Array(1.0,2.0,7.0))
randomSplitRDD(0).foreach(x => print(x +" "))
randomSplitRDD(1).foreach(x => print(x +" "))
randomSplitRDD(2).foreach(x => print(x +" "))
sc.stop　


免責聲明！

本站轉載的文章為個人學習借鑒使用，本站對版權不負任何法律責任。如果侵犯了您的隱私權益，請聯系本站郵箱yoyou2525@163.com刪除。



猜您在找 Spark函數詳解系列之RDD基本轉換 Spark函數詳解系列之RDD基本轉換 Spark常用函數講解之鍵值RDD轉換 Spark RDD詳解 spark streaming向RDD和DataFrame轉換 Spark RDD轉換為DataFrame spark轉換集合為RDD Spark基礎 --RDD算子詳解 Spark RDD和DataSet與DataFrame轉換成RDD Spark RDD概念學習系列之RDD的容錯機制（十七）
 
粵ICP備18138465號   © 2018-2025 CODEPRJ.COM