transformation

• join,leftOuterJoin,rightOuterJoin,fullOuterJoin

作用在K,V格式的RDD上。根據K進行連接，對（K,V）join(K,W)返回（K,(V,W)）

• join后的分區數與父RDD分區數多的那一個相同。

• union

合并兩個數據集。兩個數據集的類型要一致。

• 返回新的RDD的分區數是合并RDD分區數的總和。

• intersection

取兩個數據集的交集，返回新的RDD與父RDD分區多的一致

• subtract

取兩個數據集的差集，結果RDD的分區數與subtract前面的RDD的分區數一致。

• mapPartitions

與map類似，遍歷的單位是每個partition上的數據。

• distinct(map+reduceByKey+map)
• cogroup

當調用類型（K,V）和（K，W）的數據上時，返回一個數據集（K，（Iterable<V>,Iterable<W>）），子RDD的分區與父RDD多的一致。

action

• foreachPartition

遍歷的數據是每個partition的數據。

?Scala API

package ddd.henu.transformationsimport org.apache.spark.rdd.RDD import org.apache.spark.{SparkConf, SparkContext}object ElseFun {def main(args: Array[String]): Unit = {val conf = new SparkConf().setAppName("test").setMaster("local")val sc = new SparkContext(conf)sc.setLogLevel("error")/*val nameRDD: RDD[(String, Int)] = sc.parallelize(List[(String, Int)](("george", 22), ("honey", 19), ("kk", 20),("MM",18)))*///第二個參數numSlices 分區 RDD.getNumPartitionsval nameRDD: RDD[(String, Int)] = sc.parallelize(List[(String, Int)](("george", 22), ("honey", 19), ("kk", 20),("MM",18)),4)val scoreRDD: RDD[(String, Int)] = sc.parallelize(List[(String, Int)](("george", 100), ("honey", 200), ("kk", 300),("ll",400)),4)//join // val result: RDD[(String, (Int, Int))] = nameRDD.join(scoreRDD)/**結果* (george,(22,100))* (kk,(20,300))* (honey,(19,200))*///leftOuterJoin 按最大的分區/*val result: RDD[(String, (Int, Option[Int]))] = nameRDD.leftOuterJoin(scoreRDD)result.foreach(one =>{val name = one._1val v1 = one._2._1//無值的賦值10000val v2 = one._2._2.getOrElse("10000")println(s"name = $name,v1 = $v1,v2 = $v2")})*//**結果* name = george,v1 = 22,v2 = 100* name = kk,v1 = 20,v2 = 300* name = honey,v1 = 19,v2 = 200* name = MM,v1 = 18,v2 = 10000*//**結果* (george,(22,Some(100)))* (kk,(20,Some(300)))* (honey,(19,Some(200)))* (MM,(18,None))*///rightOuterJoin // val result: RDD[(String, (Option[Int], Int))] = nameRDD.rightOuterJoin(scoreRDD)/*** (ll,(None,400))* (george,(Some(22),100))* (kk,(Some(20),300))* (honey,(Some(19),200))*///fullOuterJoin // val result: RDD[(String, (Option[Int], Option[Int]))] = nameRDD.fullOuterJoin(scoreRDD)/*** (ll,(None,Some(400)))* (george,(Some(22),Some(100)))* (kk,(Some(20),Some(300)))* (honey,(Some(19),Some(200)))* (MM,(Some(18),None))*///union（合并） 8 個分區 4+4val result: RDD[(String, Int)] = nameRDD.union(scoreRDD)/*** (george,22)* (honey,19)* (kk,20)* (MM,18)* (george,100)* (honey,200)* (kk,300)* (ll,400)*/result.foreach(println)} }

?

總結

以上是生活随笔為你收集整理的Spark _06_补充部分算子【一】的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。