cogroup

官方文檔描述：

For each key k in `this` or `other`, return a resulting RDD that contains a tuple 
with the list of values for that key in `this` as well as `other`.

函數(shù)原型：

def cogroup[W](other: JavaPairRDD[K, W], partitioner: Partitioner): JavaPairRDD[K, (JIterable[V], JIterable[W])]
def cogroup[W1, W2](other1: JavaPairRDD[K, W1], other2: JavaPairRDD[K, W2],    partitioner: Partitioner): JavaPairRDD[K, (JIterable[V], JIterable[W1], JIterable[W2])]
def cogroup[W1, W2, W3](other1: JavaPairRDD[K, W1],    other2: JavaPairRDD[K, W2],    other3: JavaPairRDD[K, W3],    partitioner: Partitioner): JavaPairRDD[K, (JIterable[V], JIterable[W1], JIterable[W2], JIterable[W3])]
def cogroup[W](other: JavaPairRDD[K, W]): JavaPairRDD[K, (JIterable[V], JIterable[W])]
def cogroup[W1, W2](other1: JavaPairRDD[K, W1], other2: JavaPairRDD[K, W2]): JavaPairRDD[K, (JIterable[V], JIterable[W1], JIterable[W2])]
def cogroup[W1, W2, W3](other1: JavaPairRDD[K, W1],    other2: JavaPairRDD[K, W2],    other3: JavaPairRDD[K, W3]): JavaPairRDD[K, (JIterable[V], JIterable[W1], JIterable[W2], JIterable[W3])]
def cogroup[W](other: JavaPairRDD[K, W], numPartitions: Int): JavaPairRDD[K, (JIterable[V], JIterable[W])]
def cogroup[W1, W2](other1: JavaPairRDD[K, W1], other2: JavaPairRDD[K, W2], numPartitions: Int): JavaPairRDD[K, (JIterable[V], JIterable[W1], JIterable[W2])]
def cogroup[W1, W2, W3](other1: JavaPairRDD[K, W1],    other2: JavaPairRDD[K, W2],    other3: JavaPairRDD[K, W3],    numPartitions: Int): JavaPairRDD[K, (JIterable[V], JIterable[W1], JIterable[W2], JIterable[W3])]

源碼分析：

def cogroup[W](other: RDD[(K, W)], partitioner: Partitioner)    : RDD[(K, (Iterable[V], Iterable[W]))] = self.withScope {  
if (partitioner.isInstanceOf[HashPartitioner] && keyClass.isArray) {    
  throw new SparkException("Default partitioner cannot partition array keys.")  
}  
val cg = new CoGroupedRDD[K](Seq(self, other), partitioner)  
cg.mapValues { case Array(vs, w1s) =>    
    (vs.asInstanceOf[Iterable[V]], w1s.asInstanceOf[Iterable[W]])  
  }
}

override def getDependencies: Seq[Dependency[_]] = {  
  rdds.map { rdd: RDD[_ <: Product2[K, _]] =>    
    if (rdd.partitioner == Some(part)) {      
      logDebug("Adding one-to-one dependency with " + rdd)      
      new OneToOneDependency(rdd)    
    } else {      
      logDebug("Adding shuffle dependency with " + rdd)      
      new ShuffleDependency[K, Any, CoGroupCombiner](rdd, part, serializer)    
    }  
  }
}
override def getPartitions: Array[Partition] = {  
  val array = new Array[Partition](part.numPartitions)  
  for (i <- 0 until array.length) {    
    // Each CoGroupPartition will have a dependency per contributing RDD    
    array(i) = new CoGroupPartition(i, rdds.zipWithIndex.map { case (rdd, j) =>      
    // Assume each RDD contributed a single dependency, and get it        
    dependencies(j) match {
        case s: ShuffleDependency[_, _, _] =>          
            None        
        case _ =>          
            Some(new NarrowCoGroupSplitDep(rdd, i, rdd.partitions(i)))      
      }    
    }.toArray)  
  }  
  array
}

**
cogroup() 的計算結(jié)果放在 CoGroupedRDD 中哪個 partition 是由用戶設(shè)置的 partitioner 確定的（默認(rèn)是 HashPartitioner）。
CoGroupedRDD 依賴的所有 RDD 放進(jìn)數(shù)組 rdds[RDD] 中。再次，foreach i，如果 CoGroupedRDD 和 rdds(i) 對應(yīng)的 RDD 是 OneToOneDependency 關(guān)系，那么 Dependecy[i] = new OneToOneDependency(rdd)，否則 = new ShuffleDependency(rdd)。最后，返回與每個 parent RDD 的依賴關(guān)系數(shù)組 deps[Dependency]。
Dependency 類中的 getParents(partition id) 負(fù)責(zé)給出某個 partition 按照該 dependency 所依賴的 parent RDD 中的 partitions: List[Int]。
getPartitions() 負(fù)責(zé)給出 RDD 中有多少個 partition，以及每個 partition 如何序列化。
**

實例：

List<Integer> data = Arrays.asList(1, 2, 4, 3, 5, 6, 7, 1, 2);
JavaRDD<Integer> javaRDD = javaSparkContext.parallelize(data);

JavaPairRDD<Integer,Integer> javaPairRDD = javaRDD.mapToPair(new PairFunction<Integer, Integer, Integer>() {    
@Override    
  public Tuple2<Integer, Integer> call(Integer integer) throws Exception {        
    return new Tuple2<Integer, Integer>(integer,1);    
  }
});

//與 groupByKey() 不同，cogroup() 要 aggregate 兩個或兩個以上的 RDD。
JavaPairRDD<Integer,Tuple2<Iterable<Integer>,Iterable<Integer>>> cogroupRDD = javaPairRDD.cogroup(javaPairRDD);
System.out.println(cogroupRDD.collect());

JavaPairRDD<Integer,Tuple2<Iterable<Integer>,Iterable<Integer>>> cogroupRDD3 = javaPairRDD.cogroup(javaPairRDD, new Partitioner() {    
    @Override    
    public int numPartitions() {        
      return 2;    
    }    
    @Override    
    public int getPartition(Object key) {        
      return (key.toString()).hashCode()%numPartitions();
    }
});
System.out.println(cogroupRDD3);

join

官方文檔描述：

Return an RDD containing all pairs of elements with matching keys in `this` and `other`. 
Each* pair of elements will be returned as a (k, (v1, v2)) tuple, 
where (k, v1) is in `this` and* (k, v2) is in `other`. 
Performs a hash join across the cluster.

函數(shù)原型：

def join[W](other: JavaPairRDD[K, W]): JavaPairRDD[K, (V, W)]
def join[W](other: JavaPairRDD[K, W], numPartitions: Int): JavaPairRDD[K, (V, W)]
def join[W](other: JavaPairRDD[K, W], partitioner: Partitioner): JavaPairRDD[K, (V, W)]

源碼分析：

def join[W](other: RDD[(K, W)], partitioner: Partitioner): RDD[(K, (V, W))] = self.withScope {  
  this.cogroup(other, partitioner).flatMapValues( pair =>    
    for (v <- pair._1.iterator; w <- pair._2.iterator) yield (v, w)  
  )
}

**
從源碼中可以看出，join() 將兩個 RDD[(K, V)] 按照 SQL 中的 join 方式聚合在一起。與 intersection() 類似，首先進(jìn)行 cogroup()， 得到 <K, (Iterable[V1], Iterable[V2])> 類型的 MappedValuesRDD，然后對 Iterable[V1] 和 Iterable[V2] 做笛卡爾集，并將集合 flat() 化。
**

實例：

List<Integer> data = Arrays.asList(1, 2, 4, 3, 5, 6, 7);
final Random random = new Random();
JavaRDD<Integer> javaRDD = javaSparkContext.parallelize(data);
JavaPairRDD<Integer,Integer> javaPairRDD = javaRDD.mapToPair(new PairFunction<Integer, Integer, Integer>() {    
  @Override    
  public Tuple2<Integer, Integer> call(Integer integer) throws Exception {        
    return new Tuple2<Integer, Integer>(integer,random.nextInt(10));    
  }
});

JavaPairRDD<Integer,Tuple2<Integer,Integer>> joinRDD = javaPairRDD.join(javaPairRDD);
System.out.println(joinRDD.collect());

JavaPairRDD<Integer,Tuple2<Integer,Integer>> joinRDD2 = javaPairRDD.join(javaPairRDD,2);
System.out.println(joinRDD2.collect());

JavaPairRDD<Integer,Tuple2<Integer,Integer>> joinRDD3 = javaPairRDD.join(javaPairRDD, new Partitioner() {    
  @Override    
  public int numPartitions() {        
    return 2;    
  }    
  @Override    
  public int getPartition(Object key) {        
    return (key.toString()).hashCode()%numPartitions();
    }
});
System.out.println(joinRDD3.collect());