spark中cogroup用法
cogroup:对两个RDD中的KV元素，每个RDD中相同key中的元素分别聚合成一个集合。与reduceByKey不同的是针对两个RDD中相同的key的元素进行合并。

例一

[root@node111 ~]# spark-shell
28 一月 10:20:56 WARN [util.NativeCodeLoader] - Unable to load native-hadoop library for your platform... using builtin-java classes where applicable
Spark context Web UI available at http://node111:4040
Spark context available as 'sc' (master = spark://mycluster:7077, app id = app-20190128102110-0031).
Spark session available as 'spark'.
Welcome to
      ____              __
     / __/__  ___ _____/ /__
    _\ \/ _ \/ _ `/ __/  '_/
   /___/ .__/\_,_/_/ /_/\_\   version 2.3.1
      /_/
         
Using Scala version 2.11.8 (Java HotSpot(TM) 64-Bit Server VM, Java 1.8.0_172)
Type in expressions to have them evaluated.
Type :help for more information.
//定义集合
scala> val DBName=Array(Tuple2(1,"Spark"),Tuple2(2,"Hadoop"),Tuple2(3,"Kylin"),Tuple2(4,"Flink"))
DBName: Array[(Int, String)] = Array((1,Spark), (2,Hadoop), (3,Kylin), (4,Flink))

scala> val numType=Array(Tuple2(1,"String"),Tuple2(2,"int"),Tuple2(3,"byte"),Tuple2(4,"bollean"),Tuple2(5,"float"),Tuple2(1,"34"),Tuple2(2,"45"),Tuple2(3,"75"))
numType: Array[(Int, String)] = Array((1,String), (2,int), (3,byte), (4,bollean), (5,float), (1,34), (2,45), (3,75))
//集合并行化为RDD
scala> val names=sc.parallelize(DBName)
names: org.apache.spark.rdd.RDD[(Int, String)] = ParallelCollectionRDD[0] at parallelize at <console>:26

scala> val types=sc.parallelize(numType)
types: org.apache.spark.rdd.RDD[(Int, String)] = ParallelCollectionRDD[1] at parallelize at <console>:26
//显示原始两个RDD中的数据内容
scala> names.collect.foreach(println)
(1,Spark)
(2,Hadoop)
(3,Kylin)
(4,Flink)

scala> types.collect.foreach(println)
(1,String)
(2,int)
(3,byte)
(4,bollean)
(5,float)
(1,34)
(2,45)
(3,75)

//cogroup用法
scala> val nameAndType=names.cogroup(types)
nameAndType: org.apache.spark.rdd.RDD[(Int, (Iterable[String], Iterable[String]))] = MapPartitionsRDD[3] at cogroup at <console>:27
//显示集合内容
scala> nameAndType.collect.foreach(println)
(1,(CompactBuffer(Spark),CompactBuffer(34, String)))                            
(2,(CompactBuffer(Hadoop),CompactBuffer(45, int)))
(3,(CompactBuffer(Kylin),CompactBuffer(75, byte)))
(4,(CompactBuffer(Flink),CompactBuffer(bollean)))
(5,(CompactBuffer(),CompactBuffer(float)))

//过滤用法
scala> val commonRdd=nameAndType.filter(t=>t._2._1.iterator.hasNext && t._2._2.iterator.hasNext);
commonRdd: org.apache.spark.rdd.RDD[(Int, (Iterable[String], Iterable[String]))] = MapPartitionsRDD[4] at filter at <console>:25
//再次显示过滤后集合内容
scala> commonRdd.collect.foreach(println)
(1,(CompactBuffer(Spark),CompactBuffer(String, 34)))
(2,(CompactBuffer(Hadoop),CompactBuffer(int, 45)))
(3,(CompactBuffer(Kylin),CompactBuffer(byte, 75)))
(4,(CompactBuffer(Flink),CompactBuffer(bollean)))

scala> 

例二
spark中关于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`.
   */
  def cogroup[W](other: JavaPairRDD[K, W], partitioner: Partitioner)
  : JavaPairRDD[K, (JIterable[V], JIterable[W])] =
    fromRDD(cogroupResultToJava(rdd.cogroup(other, partitioner)))

使用的例子

private JavaPairRDD<GeoWaveInputKey, ByteArrayId> joinAndCompareTiers(
			JavaPairRDD<ByteArrayId, Tuple2<GeoWaveInputKey, Geometry>> leftTier,
			JavaPairRDD<ByteArrayId, Tuple2<GeoWaveInputKey, Geometry>> rightTier,
			Broadcast<GeomFunction> geomPredicate,
			int highestPartitionCount, 
			HashPartitioner partitioner ) {
		// Cogroup groups on same tier ByteArrayId and pairs them into Iterable
		// sets.
		JavaPairRDD<
		     ByteArrayId, 
		     Tuple2<
		             Iterable<Tuple2<GeoWaveInputKey, Geometry>>, 
		             Iterable<Tuple2<GeoWaveInputKey, Geometry>>
		      >
		> joinedTiers = leftTier
				.cogroup(
						rightTier,
						partitioner);
		
		
		// Filter only the pairs that have data on both sides, bucket strategy
		// should have been accounted for by this point.
		// We need to go through the pairs and test each feature against each
		// other
		// End with a combined RDD for that tier.
		joinedTiers = joinedTiers.filter(t -> 
				t._2._1.iterator().hasNext() &&
				t._2._2.iterator().hasNext()
		);

		
		JavaPairRDD<GeoWaveInputKey, ByteArrayId> finalMatches = joinedTiers.flatMapValues(
		(Function<
		      Tuple2<
		            Iterable<Tuple2<GeoWaveInputKey, Geometry>>, 
		            Iterable<Tuple2<GeoWaveInputKey, Geometry>>
		      >, Iterable<GeoWaveInputKey>
		>) t -> {
            GeomFunction predicate = geomPredicate.value();

            HashSet<GeoWaveInputKey> results = Sets.newHashSet();
            for (Tuple2<GeoWaveInputKey, Geometry> leftTuple : t._1) {
                for (Tuple2<GeoWaveInputKey, Geometry> rightTuple : t._2) {
                    if (predicate.call(
                            leftTuple._2,
                            rightTuple._2)) {
                        results.add(leftTuple._1);
                        results.add(rightTuple._1);
                    }
                }
            }
            return results;
        }) .mapToPair(Tuple2::swap)
        .reduceByKey(partitioner,  (id1, id2) -> id1)
        .persist(StorageLevel.MEMORY_ONLY_SER());
		
		return finalMatches;
	}