简单理解MapReudce的分区和分组

ゝ一纸荒年。 2022-08-07 17:56 70阅读 0赞

**分区**

简单的说，一组KV交给一组Reduce进行处理，分区就是制订其交付的对应规则。如下代码：

public class DoPartitioner extends HashPartitioner<MyKeyWritable, MyValueWritable>
    {
        @Override
        public int getPartition(final MyKeyWritable key, final MyValueWritable value, final int numberPartions)
        {
            return (key.getFirst().hashCode() & Integer.MAX_VALUE) % numberPartions;
        }
    }

这是Hash分区最常见的形式了，这里不再累赘。

使用上述分区方法时，发现有时候离散度不够，即有些reduce中总是分不到KV。修改为如下代码：

return (key.getFirst().hashCode() & Integer.MAX_VALUE * 127) % numberPartions;

离散度明显提高了，但是发现修改127为121后，离散度更高。其中缘由暂没有时间研究，暂时作为经验值记录下来吧。

**分组**

简单地说，分组是指定一组KV只调用一次reduce方法。社区自带的二次排序例子（SecondarySort）中，使用了分组，如下代码：

public static class FirstGroupingComparator implements RawComparator<IntPair> {
    @Override
    	public int compare(byte[] b1, int s1, int l1, byte[] b2, int s2, int l2) {
    		return WritableComparator.compareBytes(b1, s1, Integer.SIZE/8, b2, s2, Integer.SIZE/8);
    	}
    
    	@Override
    	public int compare(IntPair o1, IntPair o2) {
    		int l = o1.getFirst();
    		int r = o2.getFirst();
    		return l == r ? 0 : (l < r ? -1 : 1);
    	}
    }

示例中key是自定义的IntPair对象，包括了两个成员变量（first和second），但是分组时都只参照第一个成员变量first。

此处注意一个细节问题，假设在reduce方法中有如下代码：

for(IntWritable value: values) {
    	context.write(key, value);
    }

假设输入数据为(1,2),(1,3),(1,4)。因为具有相同first，所以对这三组KV，reduce方法只被调用一次。但是for循环中，三次的key分别是(1,2),(1,3),(1,4)，而不都是(1,2)。

如果不确定一组key是否对应一个分组，可以在分组实现代码中打上调试日志，然后在reduce日志中，应该能看到多个key调用一次reduce方法了......