Spark 1.6
以前一直模模糊糊的，现在搞一下比较清楚
combineByKey(createCombiner, mergeValue, mergeCombiners, numPartitions=None, partitionFunc=<function portable_hash at 0x7f1ac7340578>)
它是一个泛型函数，主要完成聚合操作，将输入RDD[(K,V)]转化为结果RDD[(K,C)]输出
在数据分析中，处理Key，Value的Pair数据是极为常见的场景，例如我们可以针对这样的数据进行分组、聚合或者将两个包含Pair数据的RDD根据key进行join。从函数的抽象层面看，这些操作具有共同的特征，都是将类型为RDD[(K,V)]的数据处理为RDD[(K,C)]。这里的V和C可以是相同类型，也可以是不同类型。
作者：LuciferTM
链接：http://www.jianshu.com/p/f3aea4480f2b
來源：简书
著作权归作者所有。商业转载请联系作者获得授权，非商业转载请注明出处。
流程图
多谢@蒋潇亿--知乎的回答，我这里将图用自己的话再次整理下，方便理解
复现一下代码
def createCombiner(value):
return (value,1)
def mergeValue(acc,value):
return (acc[0]+value,acc[1]+1)
def mergeCombiners(acc1,acc2):
return (acc1[0]+acc2[0],acc1[1]+acc2[1])
data = sc.parallelize([('coffee',1),('coffee',2),('panda',3),('coffee',9)],2)
# data.collect():[('coffee', 1), ('coffee', 2), ('panda', 3), ('coffee', 9)]
result = data.combineByKey(createCombiner,mergeValue,mergeCombiners)
print result.collect()
#------------------------------------------------------
#拓展，计算key所含value的均值,方法一，使用map
print result.map(lambda x:(x[0],float(x[1][0])/x[1][1])).collect()
# 方法二,s使用mapValues
print result.mapValues(lambda x:float(x[0])/x[1]).collect()
#[('coffee', (12, 3)), ('panda', (3, 1))]
#[('coffee', 4.0), ('panda', 3.0)]
#[('coffee', 4.0), ('panda', 3.0)]
来一波分析
第一种比较正统的方法，按照原理图来一步步推倒过程
如果pair RDD的key第一次出现，那么就用把该key下的value进行createCombiner操作，这里第一个pair RDD输出结果应该是这种形式
('coffee',(1,1))这里需要强调的是，这是对value进行操作的,将其中的value进行转化。这里的例子是(1,1)第一个1是value，第二个1是出现了一次对于key没重复的pair RDD才上上述同样操作，如果碰到同样key的了，那就转到第二步，key不变的情况下，将上一次的(1,1)当做参数传递进mergeValue，效果就是说，以前的acc[0]=1,即值是1，然后现在新的coffee传进来的值是2，即value=2,这样，就会对同key值进行累加acc[0]+value=同key下的累加值，而acc[1]=1，即统计了该key出现的次数，acc[1]+1=coffee这个key共出现的次数
经过这两步之后，先不考虑另一个分区的情况，如果只有一个分区，那么现在的结果应该是这样。第一个分区的结果[('coffee',(3,2)),('panda',(3,1))],然后第二个分区的结果同理('coffee',(9,1)),之后再对同key下的value传入mergeCombiner进行操作即可，方式同第二步类似，搞清楚谁是传进去的value
第二种方法，使用字典来模拟这个过程
# 相当于spark中的两个分区
part1 = [('coffee',1),('coffee',2),('panda',3)]
part2 = [('coffee',9)]
dict_res = {}
for part in [part1,part2]:
for tup in part:
if tup[0] not in dict_res:
dict_res[tup[0]]= {} # 在该key下，将value构建dict
dict_res[tup[0]]['sum'] = 0
dict_res[tup[0]]['times'] = 0
dict_res[tup[0]]['sum'] += tup[1] # sum叠加
dict_res[tup[0]]['times'] +=1 # 次数累加
print dict_res
# {'coffee': {'sum': 12, 'times': 3}, 'panda': {'sum': 3, 'times': 1}}
# 其中coffee代表键，之后的value我又传了个dict，里面key=sum的value代表和，key=times的value代表前面的key如coffee出现的次数
将上面的式子再进化一次，使更像spark的写法
def createAndMergeValue(part):
for tup in part:
if tup[0] not in dict_res:
dict_res[tup[0]]= {}
dict_res[tup[0]]['sum'] = 0
dict_res[tup[0]]['times'] = 0
dict_res[tup[0]]['sum'] += tup[1]
dict_res[tup[0]]['times'] +=1
def mergeCombiners(partitions):
for part in partitions:
createAndMergeValue(part)
dict_res = {}
partitions = [[('coffee',1),('coffee',2),('panda',3)],[('coffee',9)]] # 为了表现其分区的特性，这里用了list区分分区部分
mergeCombiners(partitions)
print dict_res
{'coffee': {'sum': 12, 'times': 3}, 'panda': {'sum': 3, 'times': 1}}
