combineByKey
官方文档描述：
Generic function to combine the elements for each key using a custom set of aggregation functions.
Turns an RDD[(K, V)] into a result of type RDD[(K, C)], for a "combined type" C Note that V and C can be different
-- for example, one might group an RDD of type (Int, Int) into an RDD of type (Int, Seq[Int]).
Users provide three functions:
- `createCombiner`, which turns a V into a C (e.g., creates a one-element list)
- `mergeValue`, to merge a V into a C (e.g., adds it to the end of a list)
- `mergeCombiners`, to combine two C's into a single one.
In addition, users can control the partitioning of the output RDD,
and whether to perform map-side aggregation (if a mapper can produce multiple items with the same key).
函数原型：
def combineByKey[C](createCombiner: JFunction[V, C], mergeValue: JFunction2[C, V, C],
mergeCombiners: JFunction2[C, C, C]): JavaPairRDD[K, C]
def combineByKey[C](createCombiner: JFunction[V, C], mergeValue: JFunction2[C, V, C],
mergeCombiners: JFunction2[C, C, C], numPartitions: Int): JavaPairRDD[K, C]
def combineByKey[C](createCombiner: JFunction[V, C], mergeValue: JFunction2[C, V, C],
mergeCombiners: JFunction2[C, C, C], partitioner: Partitioner): JavaPairRDD[K, C]
def combineByKey[C](createCombiner: JFunction[V, C], mergeValue: JFunction2[C, V, C],
mergeCombiners: JFunction2[C, C, C], partitioner: Partitioner,
mapSideCombine: Boolean,serializer: Serializer): JavaPairRDD[K, C]
**
该函数是用于将RDD[k,v]转化为RDD[k,c]，其中类型v和类型c可以相同也可以不同。
其中的参数如下：
**
- createCombiner：该函数用于将输入参数RDD[k,v]的类型V转化为输出参数RDD[k,c]中类型C;
- mergeValue：合并函数，用于将输入中的类型C的值和类型V的值进行合并，得到类型C，输入参数是（C，V），输出是C；
- mergeCombiners：合并函数，用于将两个类型C的值合并成一个类型C，输入参数是（C，C），输出是C；
- numPartitions：默认HashPartitioner中partition的个数；
- partitioner：分区函数，默认是HashPartitionner；
- mapSideCombine：该函数用于判断是否需要在map进行combine操作，类似于MapReduce中的combine，默认是 true。
源码分析：
def combineByKey[C](createCombiner: V => C,
mergeValue: (C, V) => C,
mergeCombiners: (C, C) => C,
partitioner: Partitioner,
mapSideCombine: Boolean = true,
serializer: Serializer = null): RDD[(K, C)] = self.withScope {
require(mergeCombiners != null, "mergeCombiners must be defined") // required as of Spark 0.9.0
if (keyClass.isArray) {
if (mapSideCombine) {
throw new SparkException("Cannot use map-side combining with array keys.")
}
if (partitioner.isInstanceOf[HashPartitioner]) {
throw new SparkException("Default partitioner cannot partition array keys.")
}
}
val aggregator = new Aggregator[K, V, C](
self.context.clean(createCombiner),
self.context.clean(mergeValue),
self.context.clean(mergeCombiners))
if (self.partitioner == Some(partitioner)) {
self.mapPartitions(iter => {
val context = TaskContext.get()
new InterruptibleIterator(context, aggregator.combineValuesByKey(iter, context))
}, preservesPartitioning = true)
} else {
new ShuffledRDD[K, V, C](self, partitioner)
.setSerializer(serializer)
.setAggregator(aggregator)
.setMapSideCombine(mapSideCombine)
}
}
**
从源码中可以看出，combineByKey()的实现是一边进行aggregate，一边进行compute() 的基础操作。假设一组具有相同 K 的 <K, V> records 正在一个个流向 combineByKey()，createCombiner 将第一个 record 的 value 初始化为 c （比如，c = value），然后从第二个 record 开始，来一个 record 就使用 mergeValue(c, record.value) 来更新 c，比如想要对这些 records 的所有 values 做 sum，那么使用 c = c + record.value。等到 records 全部被 mergeValue()，得到结果 c。假设还有一组 records（key 与前面那组的 key 均相同）一个个到来，combineByKey() 使用前面的方法不断计算得到 c’。现在如果要求这两组 records 总的 combineByKey() 后的结果，那么可以使用 final c = mergeCombiners(c, c’) 来计算；然后依据partitioner进行不同分区合并。
**
实例：
List<Integer> data = Arrays.asList(1, 2, 4, 3, 5, 6, 7, 1, 2);
JavaRDD<Integer> javaRDD = javaSparkContext.parallelize(data);
//转化为pairRDD
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);
}
});
JavaPairRDD<Integer,String> combineByKeyRDD = javaPairRDD.combineByKey(new Function<Integer, String>() {
@Override
public String call(Integer v1) throws Exception {
return v1 + " :createCombiner: ";
}
}, new Function2<String, Integer, String>() {
@Override
public String call(String v1, Integer v2) throws Exception {
return v1 + " :mergeValue: " + v2;
}
}, new Function2<String, String, String>() {
@Override
public String call(String v1, String v2) throws Exception {
return v1 + " :mergeCombiners: " + v2;
}
});
System.out.println("result~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~" + combineByKeyRDD.collect());
groupByKey
官方文档描述：
Group the values for each key in the RDD into a single sequence.
Allows controlling the partitioning of the resulting key-value pair RDD by passing a Partitioner.
The ordering of elements within each group is not guaranteed,
and may even differ each time the resulting RDD is evaluated.
Note: This operation may be very expensive.
If you are grouping in order to perform an aggregation (such as a sum or average) over each key,
using [[PairRDDFunctions.aggregateByKey]] or [[PairRDDFunctions.reduceByKey]] will provide much better performance.
Note: As currently implemented, groupByKey must be able to hold all the key-value pairs for any key in memory.
If a key has too many values, it can result in an [[OutOfMemoryError]].
函数原型：
def groupByKey(partitioner: Partitioner): JavaPairRDD[K, JIterable[V]]
def groupByKey(numPartitions: Int): JavaPairRDD[K, JIterable[V]]
源码分析：
def groupByKey(partitioner: Partitioner): RDD[(K, Iterable[V])] = self.withScope {
// groupByKey shouldn't use map side combine because map side combine does not
// reduce the amount of data shuffled and requires all map side data be inserted
// into a hash table, leading to more objects in the old gen.
val createCombiner = (v: V) => CompactBuffer(v)
val mergeValue = (buf: CompactBuffer[V], v: V) => buf += v
val mergeCombiners = (c1: CompactBuffer[V], c2: CompactBuffer[V]) => c1 ++= c2
val bufs = combineByKey[CompactBuffer[V]]( createCombiner, mergeValue, mergeCombiners, partitioner, mapSideCombine = false)
bufs.asInstanceOf[RDD[(K, Iterable[V])]]
}
**
从源码中可以看出groupByKey()是基于combineByKey()实现的， 只是将 Key 相同的 records 聚合在一起，一个简单的 shuffle 过程就可以完成。ShuffledRDD 中的 compute() 只负责将属于每个 partition 的数据 fetch 过来，之后使用 mapPartitions() 操作进行 aggregate，生成 MapPartitionsRDD，到这里 groupByKey() 已经结束。最后为了统一返回值接口，将 value 中的 ArrayBuffer[] 数据结构抽象化成 Iterable[]。groupByKey() 没有在 map 端进行 combine（mapSideCombine = false），这样设计是因为map 端 combine 只会省掉 partition 里面重复 key 占用的空间；但是，当重复 key 特别多时，可以考虑开启 combine。
**
实例：
List<Integer> data = Arrays.asList(1, 2, 4, 3, 5, 6, 7);
JavaRDD<Integer> javaRDD = javaSparkContext.parallelize(data);
//转为k，v格式
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);
}
});
JavaPairRDD<Integer,Iterable<Integer>> groupByKeyRDD = javaPairRDD.groupByKey(2);
System.out.println(groupByKeyRDD.collect());
//自定义partition
JavaPairRDD<Integer,Iterable<Integer>> groupByKeyRDD3 = javaPairRDD.groupByKey(new Partitioner() {
//partition各数
@Override
public int numPartitions() { return 10; }
//partition方式
@Override
public int getPartition(Object o) {
return (o.toString()).hashCode()%numPartitions();
}
});
System.out.println(groupByKeyRDD3.collect());
