轉換

含義

map(func)

返回一個新的RDD，該RDD由每一個輸入元素經過func函式轉換後組成

filter(func)

返回一個新的RDD，該RDD由經過func函式計算後返回值為true的輸入元素組成

flatMap(func)

先map，再flatten壓平

union(otherDataset)

對源RDD和引數RDD求並集後返回一個新的RDD

intersection(otherDataset)

對源RDD和引數RDD求交集後返回一個新的RDD

subtract(otherDataset)

求差集後返回新的RDD，出現在源rdd中，不在otherrdd中

distinct([numTasks]))

對源RDD進行去重後返回一個新的RDD

mapPartitions(func)

類似於map，但獨立地在RDD的每一個分片上執行，因此在型別為T的RDD上執行時，func的函式型別必須是Iterator[T] => Iterator[U]

mapPartitionsWithIndex(func)

類似於mapPartitions，但func帶有一個整數引數表示分片的索引值，因此在型別為T的RDD上執行時，func的函式型別必須是

(Int, Interator[T]) => Iterator[U]

sortBy(func,[ascending], [numTasks])

與sortByKey類似，但是更靈活

sortByKey([ascending], [numTasks])

在一個(K,V)的RDD上呼叫，K必須實現Ordered介面，返回一個按照key進行排序的(K,V)的RDD

join(otherDataset, [numTasks])

在型別為(K,V)和(K,W)的RDD上呼叫，返回一個相同key對應的所有元素對在一起的(K,(V,W))的RDD

cogroup(otherDataset, [numTasks])

在型別為(K,V)和(K,W)的RDD上呼叫，返回一個(K,(Iterable<V>,Iterable<W>))型別的RDD

cartesian(otherDataset)

笛卡爾積

mapValues(func)

在一個(K,V)的RDD上呼叫

groupBy (func, [numTasks])

根據自定義條件進行分組

groupByKey([numTasks])

在一個(K,V)的RDD上呼叫，返回一個(K, Iterator[V])的RDD

reduceByKey(func, [numTasks])

在一個(K,V)的RDD上呼叫，返回一個(K,V)的RDD，使用指定的reduce函式，將相同key的值聚合到一起，與groupByKey類似，reduce任務的個數可以通過第二個可選的引數來設定

aggregateByKey(zeroValue)(seqOp, combOp, [numTasks])

針對分割槽內部使用seqOp方法，針對最後的結果使用combOp方法。

coalesce(numPartitions)      

用於對RDD進行重新分割槽，第一個引數是分割槽的數量，第二個引數是是否進行shuffle，可不傳，預設不shuffle

repartition(numPartitions)

用於對RDD進行重新分割槽，相當於shuffle版的calesce

動作

含義

reduce(func)

通過func函式聚集RDD中的所有元素

collect()

在驅動程式中，以陣列的形式返回資料集的所有元素

collectAsMap

類似於collect。該函式用於Pair RDD，最終返回Map型別的結果。

count()

返回RDD的元素個數

first()

返回RDD的第一個元素（類似於take(1)）

take(n)

返回一個由資料集的前n個元素組成的陣列

saveAsTextFile(path)

將資料集的元素以textfile的形式儲存到HDFS檔案系統或者其他支援的檔案系統

top（n）

按照預設排序（降序） 取資料

takeOrdered(n, [ordering])

與top類似，順序相反  預設是升序

countByKey()

針對(K,V)型別的RDD，返回一個(K,Int)的map，表示每一個key對應的元素個數。

foreach(func)

在資料集的每一個元素上，執行函式func進行更新。foreach，任務在executor中執行，列印資訊也會在executor中顯示

foreachPartition

對分割槽進行操作

rdd1.foreach(println) rdd1.foreachPartition(it=>println(it.mkString("")))

// 重分割槽的api     rdd1.coalesce(2)     // coalesce(numPartitions: Int, shuffle: Boolean = false     rdd1.repartition(2)  // coalesce(numPartitions, shuffle = true) //    repartition 就是coalesce,第二個引數為true     /**  repartition 要進行資料的shuffle   不管是擴大分割槽，還是減少分割槽，都進行shuffle       *  coalesce預設沒有進行資料的shuffle   減少分割槽，直接是分割槽合併。       *  coalesce   擴大分割槽,        */

val f = (i:Int,it:Iterator[Int]) =>

     it.map(t=> s"part:$i,values:$t")

def reduceByKey(func: (V, V) => V, numPartitions: Int):

/**       * map       * mapValues       * mapPartition      操作的是每一個分割槽    函式的輸入引數型別是Iterator[元素型別]       * mapPartitionWithIndex       *       */     // 建立rdd  同時指定分割槽的數量為2個  資料會被打散，然後平均分配     val rdd = sc.makeRDD(List(1, 3, 5, 7, 9), 2) // part 0 1     rdd.map({       i =>         // 具體的元素         i * 10     })     // 該方法每次操作的物件是一個迭代器，對應的就是一個分割槽的資料     rdd.mapPartitions({       it =>         // 分割槽         it.map(_ * 10)     })     // 函式     val f=(index:Int,it:Iterator[Int])=>{       it.map({         t=> s"part:$index,value=$t"       })     }     val index1: RDD[String] = rdd.mapPartitionsWithIndex({       // 第一個引數，是分割槽的索引  分割槽編號       // 第二個引數： 分割槽的資料 Iterator[Int]       case (index, it) =>         it.map({           t =>             s"part:$index,value=$t"         })     })//    ArrayBuffer(part:0,value=1, part:0,value=3, part:1,value=5, part:1,value=7, part:1,value=9)     // 收集資料並列印     println(index1.collect().toBuffer)