淘宝数据仓库团队，在“HADOOP中一种非典型两表JOIN的处理方法”这篇文章里， 无私地 share 了他们的方法。我就着他们的写一个续，权当讨论，抛砖引玉。淘宝团队的文章主要说的是大规模数据情况下如何计算，我这篇接着他们最后的问题，即“多对多”的情况说一下思路。

要解决的问题可以简化描述一下：

1. 有两组数据，input1 { P1, U1; P1, U2; P2, U3; P3, U4; P4, U4 }，input2 { P1, C1; P1, C2; P2, C3; P3, C3; P3, C4; P4, C4 }。
2. 要求执行类似于数据库两表 Inner Join 的操作，以 P 为 key，建立起 U 和 C 直接的对应关系，即最终结果为 output { U1, C1; U1, C2; U2, C1; U2, C2; U3, C3; U4, C3; U4, C4 }。

在数据库里，使用类似的 SQL 可以达到要求：SELECT DISTINCT(U, C) FROM input1 INNER JOIN input2 ON  input1.P=input2.P。但如果要放在 Hadoop 里面求解，就需要动些脑筋了。

研究这个问题，首先需要理解 Hadoop 的运行机制。简单来讲，Hadoop 分为 Map 和 Reduce 两个操作：Map 操作将输入（如一行数据）格式化为 <key: value1><key: value2><key: value3> … <key: valueN>这样的一组结果，作为 Map 的输出。Hadoop 在 Map 和 Reduce 之间，会自动把 Map 的输出按照 key 合并起来，作为 Reduce 的输入。Reduce 得到这样一个 {key: [value1, value2, value3, ..., valueN]} 的输入之后，就可以进行自己的处理，完成最终计算了。

针对于我们这里要解决的问题，步骤如下。

1. 将 Map 的输入构造为下面的格式：来自于 input1 的输入格式化为 {<input1, P1>: U1, U2}；来自于 input2 的输入格式化为 {<input2, P1>: C1, C2}。
2. 在 Map 操作内，将数据转化为 {P1: <input1, U1>}，{P1: <input1, U2>}，{P1: <input2, C1>}，{P1: <input2, C2>}，作为 Reduce 操作的输入。
3. 经过 Hadoop 内部自己的操作，实际 Reduce 操作的输入为：{P1: <input1, U1>, <input1, U2>, <input2, C1>, <input2, C2>}。
4. Reduce 里操作会复杂一下。首先需要执行一次 regroup，得到如下的结果 {<input1>: <input1, U1>, <input1, U2>; <input2>: <input2, C1>, <input2, C2>}。把这个结果拆开，可以得到两个集合：{<input1>, <input2>} 与 {[<input1, U1>, <input1, U2>], [<input2, C1>, <input2, C2>]}。
5. 循环集合2，即可以得到最终结果。不过在 Reduce 里面作这个循环是需要一定技巧的，讲起来比较绕，大家就直接看后面的代码吧。
6. 在此 Reduce 的结果之上，再跑一个 Map/Reduce，还可以得到 <U, C>的次数，作为每个组合的权重。

对于大数据量，需要启用 Hadoop 的数据压缩功能。

这是一个通用地解决 Inner Join 问题的思路，在 Hadoop 的 contrib package 里有具体的代码实现，参见 org.apache.hadoop.contrib.utils.join。
国内还有哪个 team 在用 Hadoop？欢迎交流！