Hadoop之combiner和partitioner

1. Combiner

通常，每一个map可能会产生大量的输出，combiner的作用就是在map端对输出先做一次合并，以减少传输到reducer的数据量。

我们以计算特定key对应值的平均值为例，展示一下combiner的用法：

class Mapper
    method Map(string t, integer r)
    Emit(string t, integer r)

class Combiner
    method Combine(string t, integers [r1, r2, . . .])
    sum ← 0
    cnt ← 0
    for all integer r ∈ integers [r1, r2, . . .] do
        sum ← sum + r
        cnt ← cnt + 1
    Emit(string t, pair (sum, cnt))  // Separate sum and count

class Reducer
    method Reduce(string t, pairs [(s1, c1), (s2, c2) . . .])
    sum ← 0
    cnt ← 0
    for all pair (s, c) ∈ pairs [(s1, c1), (s2, c2) . . .] do
        sum ← sum + s
        cnt ← cnt + c
    ravg ← sum/cnt
    Emit(string t, integer ravg)

乍一看应该没有问题，但是不幸的是，这个combiner是不正确的。因为框架要求，combiner的输入输出类型必须和mapper的输出以及reducer的输入类型一致。而上面的伪代码中，mapper的输出类型为<string, integer>，而combiner的输出类型为<string, pair<integer, integer>>，这样的话，combiner就不能正常工作。

改变的方式非常简单，把mapper的输出包装一下即可：

class Mapper
    method Map(string t, integer r)
        Emit(string t, pair (r, 1))

class Combiner
    method Combine(string t, pairs [(s1, c1), (s2, c2) . . .])
    sum ← 0
    cnt ← 0
    for all pair (s, c) ∈ pairs [(s1, c1), (s2, c2) . . .] do
        sum ← sum + s
        cnt ← cnt + c
    Emit(string t, pair (sum, cnt))

class Reducer
    method Reduce(string t, pairs [(s1, c1), (s2, c2) . . .])
    sum ← 0
    cnt ← 0
    for all pair (s, c) ∈ pairs [(s1, c1), (s2, c2) . . .] do
        sum ← sum + s
        cnt ← cnt + c
    ravg ← sum/cnt
    Emit(string t, integer ravg)

2. Partitioner

首先需要继承自Partitioner类（在0.19中为Partitioner接口），并重载它的getPartition方法：

然后在job配置中设置Partitioner Class：