Pair RDD: 提供並行操作各個鍵和跨節點重新進行資料分組的操作介面

建立Pair RDD

1) 把普通的RDD轉化為Pair RDD

使用map方法將lines劃分為以首個單詞為鍵，行內容為值的Pair RDD

val pairs = lines.map(x => (x.split(" ")(0), x)

2) 驅動器程式中建立Pair RDD

呼叫SparkContext.parallelize()

val pairs = sc.parallelize(List((1,1), (2, 2), (3, 3)))

轉化操作

1 Pair RDD也是RDD，所以適用於普通RDD的函式也適用於Pair RDD，如filter

對Pair RDD過濾出長度小於20的鍵值對

pairs.filter{case(key, value) => value.length < 20}

2 聚合操作

> reduceByKey()：接收一個函式，並使用該函式對值進行合併。為資料集的每個鍵進行並行的規約操作，每個規約操作會將鍵相同的值合併起來。返回鍵和規約結果組成的新的RDD

> foldByKey()：使用一個與RDD和合並函式中的資料類性相同的零值作為初始值。對零值和另一個元素合併，結果仍是該元素

> 使用reduceByKey和mapValues()計算每個鍵的對應值的均值

rdd.mapValues(x => (x, 1)).reduceByKey((x, y) => (x._1 + y._1, x._2 + y._2))
 

> reduceByKey單詞計數

val input = sc.textFile("hdfs://...")

val words = input.flatMap(x => x.split(" "))

val result = words.map(x => (x, 1)).reduceByKey((x, y) => x + y)

> combineByKey()最為常用的基於鍵的聚合函式，讓使用者返回與輸入資料型別不同的返回值

    > combineByKey()使用一個叫createCombiner()函式建立鍵對應的累加器初始值，這個操作是對分割槽的，不是對整個RDD的

    > 如果處理當前分割槽已經初始化的鍵，則使用mergeValue()方法將累加值合併

    > 每個分割槽是獨立處理，一個鍵可能有多個累加器，如果要合併不同分割槽的累加器，使用mergeCombiners()

> 自定義並行度

   val data = Seq(("a", 3), ("b", 4), ("a", 1))

   sc.parallelize(data).reduceByKey((x, y) => x + y) //預設並行度

   sc.parallelize(data).reduceByKey((x, y) => x + y, 10) //自定義並行度

也可使用repartition()和coalesce()方法對資料混洗

資料分組

單個RDD分組：groupByKey()將資料根據鍵分組，RDD(K, V)轉為RDD(K, Iterable(V))

多個共享一個鍵的RDD進行分組，RDD(K, V)和RDD(K, M)轉為RDD(K, Iterable(V),Iterable(M))

連線

Join() rightOuterJoin() leftOuterJoin()

資料排序

sortByKey()

行動操作