符号、记法

其中{k,v}指一个Key,Value对，{..} 中第一个分量是Key，第二个是Value。

[e]指一个集合，其中的元素为e。 [{k,v}]就指一个{k,v}的集合。

问题

给定巨大的集合S=[{k1,k2}]，对S中每个k1，计算k1相同，而k2不同的元素个数。生成[{k1,distinct_count([k2])}]。

其中S数据的特点是：

1.         不同k1的数目很少（少于1K，很可能小于机器数目）；

2.         不同k2的数目很大（大于20M）；

3.         总的记录数目很大（大于1G）；

4.         k1的分布很不均匀，最频繁的k1，其记录数占到总数的1/3，对应的独立k2几乎包括整个k2集合；

5.         k2的分布可以认为是均匀分布。

数据集的这些特点，也正是这种方法的适用条件。如果k1,k2的分布都很均匀，并且不同k1的数目远大于机器数目，还是用传统方案比较简单。

MapReduce解决方案

传统方案

Map.Input={k1,k2}

Map.Output={k1,k2}

Partition=Hash(k1)

Reduce.Input={k1,k2}

Reduce.Output={k1,distinct_count(k2)}

Reduce中，对每个k1，需要做set_union，计算
distinct_count(k2)=count(set_union([[k2]]))

这种方法的问题是因为k1的分布不均匀，导致负载不平衡，最频繁的k1，成为计算的瓶颈，无法提高并行度。

倒转Key

Map1.Input={k1,k2}

Map1.Output={k2,k1}

Partition1=Hash(k2)

Reduce1.Input=Map1.Output={k2,k1}

Reduce1.Output={k1,cnt=distinct_count([{k1,k2}])}

Map2.Input=Reduce1.Output

Map2=Identity

Partition2=Hash(k1)

Reduce2.Input=Map2.Output={k1,cnt}

Reduce2.Output={k1,sum([cnt])}

l  在Map1和Reduce1之间，按Hash(k2)进行partition。

l  每个Reduce1处理的数据就只包含k2的一个子集；因为k2均匀分布，每个子集尺寸相当；最关键之处就在这里，这样一个转换，达到了负载均衡！

l  Reduce1维护一个全局的hash_map<k1,hash_set<k2> >，对每条记录，分别进行一次k1和k2的查找或插入【m[k1].insert(k2)】。

l  Reduce2每读一条输入，要进行一次k1的查找，并累加该k1->cnt【m[k1]+=cnt】。