Spark運算元：transformation之union、intersection、subtract
阿新 • • 發佈：2018-12-11
1、union：def union(other: RDD[T]): RDD[T]
該函式是不去重的合併兩個RDD。
scala> var rdd1 = sc.makeRDD(1 to 2,1) rdd1: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[45] at makeRDD at :21 scala> rdd1.collect res42: Array[Int] = Array(1, 2) scala> var rdd2 = sc.makeRDD(2 to 3,1) rdd2: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[46] at makeRDD at :21 scala> rdd2.collect res43: Array[Int] = Array(2, 3) //union合併 scala> rdd1.union(rdd2).collect res44: Array[Int] = Array(1, 2, 2, 3)
2、intersection：
def intersection(other: RDD[T]): RDD[T] def intersection(other: RDD[T], numPartitions: Int): RDD[T] def intersection(other: RDD[T], partitioner: Partitioner)(implicit ord: Ordering[T] = null): RDD[T]
該函式是取兩個RDD的交集，且去重。numPartitions表示分割槽數，partitioner表示分割槽函式。
scala> var rdd1 = sc.makeRDD(1 to 2,1) rdd1: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[45] at makeRDD at :21 scala> rdd1.collect res42: Array[Int] = Array(1, 2) scala> var rdd2 = sc.makeRDD(2 to 3,1) rdd2: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[46] at makeRDD at :21 scala> rdd2.collect res43: Array[Int] = Array(2, 3) //intersection交集 scala> rdd1.intersection(rdd2).collect res45: Array[Int] = Array(2) scala> var rdd3 = rdd1.intersection(rdd2) rdd3: org.apache.spark.rdd.RDD[Int] = MapPartitionsRDD[59] at intersection at :25 scala> rdd3.partitions.size res46: Int = 1 scala> var rdd3 = rdd1.intersection(rdd2,2) rdd3: org.apache.spark.rdd.RDD[Int] = MapPartitionsRDD[65] at intersection at :25 scala> rdd3.partitions.size res47: Int = 2
3、subtract：
def subtract(other: RDD[T]): RDD[T] def subtract(other: RDD[T], numPartitions: Int): RDD[T] def subtract(other: RDD[T], partitioner: Partitioner)(implicit ord: Ordering[T] = null): RDD[T]
該函式是取兩個RDD的差集，不去重。
scala> var rdd1 = sc.makeRDD(Seq(1,2,2,3)) rdd1: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[66] at makeRDD at :21 scala> rdd1.collect res48: Array[Int] = Array(1, 2, 2, 3) scala> var rdd2 = sc.makeRDD(3 to 4) rdd2: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[67] at makeRDD at :21 scala> rdd2.collect res49: Array[Int] = Array(3, 4) scala> rdd1.subtract(rdd2).collect res50: Array[Int] = Array(1, 2, 2)