pyspark中combineByKey的两种理解方法

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)]

---

来一波分析

第一种比较正统的方法，按照原理图来一步步推倒过程

1. 如果pair RDD的key第一次出现，那么就用把该key下的value进行createCombiner操作，这里第一个pair RDD输出结果应该是这种形式('coffee',(1,1))这里需要强调的是，这是对value进行操作的,将其中的value进行转化。这里的例子是(1,1)第一个1是value，第二个1是出现了一次

2. 对于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共出现的次数

3. 经过这两步之后，先不考虑另一个分区的情况，如果只有一个分区，那么现在的结果应该是这样。第一个分区的结果[('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}}

---

致谢

@转--pyspark-combineByKey详解