hadoop job解决大数据量关联的一种方法

http://www.geminikwok.com/2011/04/02/hadoop-job%C3%A8%C2%A7%C2%A3%C3%A5%C2%86%C2%B3%C3%A5%C2%A4%C2%A7%C3%A6%C2%95%C2%B0%C3%A6%C2%8D%C2%AE%C3%A9%C2%87%C2%8F%C3%A5%C2%85%C2%B3%C3%A8%C2%81%C2%94%C3%A7%C2%9A%C2%84%C3%A4%C2%B8%C2%80%C3%A7%C2%A7%C2%8D%C3%A6%C2%96%C2%B9%C3%A6%C2%B3%C2%95/

用java写的hadoop程序实现关联，首先重写分发类（PKPartitioner）和比较类（PVComparator）：

 

public static class PKPartitioner<K2 extends WritableComparable<Text>, V2 extends Writable>
extends HashPartitioner<K2, V2> 
{

public PKPartitioner() {}

public int getPartition(K2 key, V2 value, int numReduceTasks) {

return (key.toString().split(“\u0002″)[0].hashCode() & Integer.MAX_VALUE) % numReduceTasks;}

}


public static class PVComparator extends WritableComparator {

public PVComparator() {super(Text.class, true);}

private int compareStr(String a, String b) {

String aTokens = a.split(“\u0002″)[0];

String bTokens = b.split(“\u0002″)[0];

return aTokens.compareTo(bTokens);}

public int compare(Object a, Object b) {

String aStr = a.toString();

String bStr = b.toString();

int i = compareStr(aStr, bStr);

return i;}

public int compare(WritableComparable a, WritableComparable b) {

String aStr = a.toString();

String bStr = b.toString();

return compareStr(aStr, bStr);}

}

 

实现思想是，在map输出的key后面加入一个非常用字符（^A），在这个字符之后加上两个不同的字符（用字符序的一大一小，来形成reduce节点上的数据排序，字母序小的排前面）。map分发数据时，以^A前面的原串为标准，在重写的compare函数的配合下相同的key还是会到同一个reduce节点上，以此实现大数据量数据的关联。

PKPartitioner这个类是用来决定map输出时，什么样的key输出到同一个reduce节点（调用setPartitionerClass方法），但不保证到同一个reduce节点的key会在同一个group（即不一定在reduce的同一个iter迭代里）。注：map端当输出buffer到达内存的一定比例时，将内存中的数据写到磁盘，此时会按key进行排序，然后才写入磁盘。

由于PKPartitioner不保证同一个reduce节点的key会到一个group中，所以还需要PVComparator 来指定group的划分方法（调用setOutputValueGroupingComparator方法），一个group即一个iter迭代。

另外，还有一个setOutputKeyComparatorClass方法，这个很容易和setOutputValueGroupingComparator混淆，setOutputValueGroupingComparator是用来觉得group划分，而setOutputKeyComparatorClass是用来决定key的排序。默认情况下，reduce节点上的数据是按key的升序字母序来排的，如果你要改变这个顺序（比如数字序）可以用setOutputKeyComparatorClass来指定自定义的类。