MapReduce分组

•  分组：相同key的value进行分组

 例子：如下输入输出，右边的第一列没有重复值，第二列取得是当第一列相同时第二例取最大值

          

分析：首先确定<k3,v3>,k3的选择两种方式，

方法1.前两列都作为k3

方法2.两列分别是k3和v3，此种情况的k2和v2分别是那些，第一列为k2，第二列为v2，但是最后如何无法转化为k3，v3呢，思路是从v2s中取值最大的，此种情况不能取值。

第一部分：方法二达到任务目的

(1)自定义Mapper

private static class MyMapper extends Mapper<LongWritable, Text, IntWritable, IntWritable>{
    IntWritable k2= new IntWritable();
    IntWritable v2= new IntWritable();
    @Override
    protected void map(LongWritable key, Text value,
            Mapper<LongWritable, Text, IntWritable, IntWritable>.Context context)
            throws IOException, InterruptedException {
           String[] splited = value.toString().split("	");
           k2.set(Integer.parseInt(splited[0]));
           v2.set(Integer.parseInt(splited[1]));
           context.write(k2, v2);
    }
}

(2)自定义Reduce

//按照k2進行排序，分組（分为3各组，reduce函数被调用3次，分几组调用几次）
//分组为3-{3,2,1}, 2-{2,1},1-{1}

private static class MyReducer extends Reducer<IntWritable, IntWritable, IntWritable, IntWritable>{
    IntWritable v3 = new IntWritable();
    @Override
    protected void reduce(IntWritable k2, Iterable<IntWritable> v2s,
            Reducer<IntWritable, IntWritable, IntWritable, IntWritable>.Context context)
            throws IOException, InterruptedException {
        int max=Integer.MIN_VALUE;
        for (IntWritable v2 : v2s) {
            if (v2.get()>max) {
                max=v2.get();
            }
        }
        //每个组求得一个最大值可得到结果的序列
        v3.set(max);
        context.write(k2, v3);
    }
}

(3)组合MapReduce

public static void main(String[] args) throws Exception {
    Job job = Job.getInstance(new Configuration(), GroupTest.class.getSimpleName());
    job.setJarByClass(GroupTest.class);
    //1.自定义输入路径
    FileInputFormat.setInputPaths(job, new Path(args[0]));
    //2.自定义mapper
    //job.setInputFormatClass(TextInputFormat.class);
    job.setMapperClass(MyMapper.class);
    //job.setMapOutputKeyClass(Text.class);
    //job.setMapOutputValueClass(TrafficWritable.class);
    
    //3.自定义reduce
    job.setReducerClass(MyReducer.class);
    job.setOutputKeyClass(IntWritable.class);
    job.setOutputValueClass(IntWritable.class);
    //4.自定义输出路径
    FileOutputFormat.setOutputPath(job, new Path(args[1]));
    //job.setOutputFormatClass(TextOutputFormat.class);//对输出的数据格式化并写入磁盘
    
    job.waitForCompletion(true);
}

由此，完整的代码如下：

package Mapreduce;

import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

public class GroupTest {
public static void main(String[] args) throws Exception {
    Job job = Job.getInstance(new Configuration(), GroupTest.class.getSimpleName());
    job.setJarByClass(GroupTest.class);
    //1.自定义输入路径
    FileInputFormat.setInputPaths(job, new Path(args[0]));
    //2.自定义mapper
    //job.setInputFormatClass(TextInputFormat.class);
    job.setMapperClass(MyMapper.class);
    //job.setMapOutputKeyClass(Text.class);
    //job.setMapOutputValueClass(TrafficWritable.class);
    
    //3.自定义reduce
    job.setReducerClass(MyReducer.class);
    job.setOutputKeyClass(IntWritable.class);
    job.setOutputValueClass(IntWritable.class);
    //4.自定义输出路径
    FileOutputFormat.setOutputPath(job, new Path(args[1]));
    //job.setOutputFormatClass(TextOutputFormat.class);//对输出的数据格式化并写入磁盘
    
    job.waitForCompletion(true);
}
private static class MyMapper extends Mapper<LongWritable, Text, IntWritable, IntWritable>{
    IntWritable k2= new IntWritable();
    IntWritable v2= new IntWritable();
    @Override
    protected void map(LongWritable key, Text value,
            Mapper<LongWritable, Text, IntWritable, IntWritable>.Context context)
            throws IOException, InterruptedException {
           String[] splited = value.toString().split("	");
           k2.set(Integer.parseInt(splited[0]));
           v2.set(Integer.parseInt(splited[1]));
           context.write(k2, v2);
    }
}
//按照k2進行排序，分組（分为3各组，reduce函数被调用3次，分几组调用几次）
//分组为3-{3,2,1}, 2-{2,1},1-{1}
private static class MyReducer extends Reducer<IntWritable, IntWritable, IntWritable, IntWritable>{
    IntWritable v3 = new IntWritable();
    @Override
    protected void reduce(IntWritable k2, Iterable<IntWritable> v2s,
            Reducer<IntWritable, IntWritable, IntWritable, IntWritable>.Context context)
            throws IOException, InterruptedException {
        int max=Integer.MIN_VALUE;
        for (IntWritable v2 : v2s) {
            if (v2.get()>max) {
                max=v2.get();
            }
        }
        //每个组求得一个最大值可得到结果的序列
        v3.set(max);
        context.write(k2, v3);
    }
}
}

(4)测试代码运行结果

•   [root@neusoft-master filecontent]# hadoop jar GroupTest.jar /neusoft/twoint  /out9 
•   [root@neusoft-master filecontent]# hadoop jar -text  /out9/part-r-00000
•   [root@neusoft-master filecontent]# hadoop dfs -text  /out9/part-r-00000  

       

第二部分：方法一达到任务目的

      前两列都作为k3，无v3，由此类推，k2也是前两列

      但是如果采用默认分组的话，上述数据集分为6组，无法达到同样的数值取得最大值的目的。

      由此，利用Mapreduce的自定义分组规则，使得第一列相同的数值可以在一个组里面，从而正确的分组。

      MapReduce提供了job.setGroupingComparatorClass(cls);其中cls是自定义分组的类

      

      (1) 从源代码可知，该类需要继承RawComparator类，自定义分组代码如下：

//分组比较--自定义分组
    private static class MyGroupingComparator implements RawComparator {
        public int compare(Object o1, Object o2) {
            return 0;//默认的比较方法
        }
        //byte[] b1 表示第一个参数的输入字节表示，byte[] b2表示第二个参数的输入字节表示
        //b1 The first byte array. 第一个字节数组，
        //b1表示前8个字节，b2表示后8个字节，字节是按次序依次比较的
        //s1 The position index in b1. The object under comparison's starting index.第一列开始位置
        //l1 The length of the object in b1.第一列长度 ，在这里表示长度8
        //提供的数据集中的k2一共48个字节，k2的每一行的TwoInt类型表示8字节（t1和t2分别为4字节）
        public int compare(byte[] b1, int s1, int l1, byte[] b2, int s2, int l2) {
            //compareBytes是按字节比较的方法，其中k2表示的是两列，第一列比较，第二例不比较
            //第一个字节数组的前四个字节和第二个字节数组的前四个字节比较
            //{3,3}，{3,2},{3,1},{2,2},{2,1},{1,1}
            //比较上述六组的每组的第一个数字，也就是比较twoint中得t1数值
            //现在就根据t1可分成3个组了{3,(3,2,1)}{2,(2,1)}{1,1}
            //之后再从v2中取出最大值
            return WritableComparator.compareBytes(b1, s1, l1-4, b2, s2, l2-4);
        }

    }

      (2)主函数中调用      

//当没有下面的自定义分组的话，会调用k2的compareto方法执行k2的比较，如果自定义了分组类则使用自定义分组类
        job.setGroupingComparatorClass(MyGroupingComparator.class);

     (3)根据比较函数个字段的含义，可以得到v2的类型为intwritable，而不是nullwritable，v2是由第二列的数组成的集合

      Mapper函数如下：

private static class MyMapper extends
            Mapper<LongWritable, Text, TwoInt, IntWritable> {
        //这里的v2需要改为IntWritable而不是nullwritable
        TwoInt K2 = new TwoInt();
        IntWritable v2= new IntWritable();
        @Override
        protected void map(LongWritable key, Text value,
                Mapper<LongWritable, Text, TwoInt, IntWritable>.Context context)
                throws IOException, InterruptedException {
            String[] splited = value.toString().split("	");
            K2.set(Integer.parseInt(splited[0]), Integer.parseInt(splited[1]));
            v2.set(Integer.parseInt(splited[1])); //要比较第二列，需要将第二列的值赋值为v2
            context.write(K2, v2);
        }
    }

     (4)k3和v3的类型为reduce输出的类型，均为intwritable类型，但是如何根据得到的v2去统计其中相同key的value中得最大值呢？

private static class MyReducer extends
            Reducer<TwoInt, IntWritable, IntWritable, IntWritable> {//k2,v2s,k3,v3
        IntWritable k3 = new IntWritable();
        IntWritable v3 = new IntWritable();
        @Override
        protected void reduce(
                TwoInt k2,
                Iterable<IntWritable> v2s,
                Reducer<TwoInt, IntWritable, IntWritable, IntWritable>.Context context)
                throws IOException, InterruptedException {
            int max=Integer.MIN_VALUE;
            for (IntWritable v2 : v2s) {
                if (v2.get()>max) {
                    max=v2.get();
                }
            }
            //每个组求得一个最大值可得到结果的序列
            v3.set(max);
            k3.set(k2.t1);//k2的第一列作为k3,因为k2为Twoint类型
            context.write(k3,v3);
        }
    }

最终的代码如下：

package Mapreduce;

import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.RawComparator;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.WritableComparable;
import org.apache.hadoop.io.WritableComparator;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;


public class Group2Test {
    public static void main(String[] args) throws Exception {
        Job job = Job.getInstance(new Configuration(),
                Group2Test.class.getSimpleName());
        job.setJarByClass(Group2Test.class);
        // 1.自定义输入路径
        FileInputFormat.setInputPaths(job, new Path(args[0]));
        // 2.自定义mapper
        job.setMapperClass(MyMapper.class);
        //这里的k2,v2和k3,v3不同,需要显式定义k2和v2类型
        job.setMapOutputKeyClass(TwoInt.class);  
        job.setMapOutputValueClass(IntWritable.class);

        //当没有下面的自定义分组的话，会调用k2的compareto方法执行k2的比较，如果自定义了分组类则使用自定义分组类
        job.setGroupingComparatorClass(MyGroupingComparator.class);

        // 3.自定义reduce
        job.setReducerClass(MyReducer.class);
        job.setOutputKeyClass(IntWritable.class);
        job.setOutputValueClass(IntWritable.class);
        // 4.自定义输出路径
        FileOutputFormat.setOutputPath(job, new Path(args[1]));

        job.waitForCompletion(true);
    }
    //分组比较--自定义分组
    private static class MyGroupingComparator implements RawComparator {
        public int compare(Object o1, Object o2) {
            return 0;//默认的比较方法
        }
        //byte[] b1 表示第一个参数的输入字节表示，byte[] b2表示第二个参数的输入字节表示
        //b1 The first byte array. 第一个字节数组，
        //b1表示前8个字节，b2表示后8个字节，字节是按次序依次比较的
        //s1 The position index in b1. The object under comparison's starting index.第一列开始位置
        //l1 The length of the object in b1.第一列长度 ，在这里表示长度8
        //提供的数据集中的k2一共48个字节，k2的每一行的TwoInt类型表示8字节（t1和t2分别为4字节）
        public int compare(byte[] b1, int s1, int l1, byte[] b2, int s2, int l2) {
            //compareBytes是按字节比较的方法，其中k2表示的是两列，第一列比较，第二例不比较
            //第一个字节数组的前四个字节和第二个字节数组的前四个字节比较
            //{3,3}，{3,2},{3,1},{2,2},{2,1},{1,1}
            //比较上述六组的每组的第一个数字，也就是比较twoint中得t1数值
            //现在就根据t1可分成3个组了{3,(3,2,1)}{2,(2,1)}{1,1}
            //之后再从v2中取出最大值
            return WritableComparator.compareBytes(b1, s1, l1-4, b2, s2, l2-4);
        }

    }

    private static class MyMapper extends
            Mapper<LongWritable, Text, TwoInt, IntWritable> {
        //这里的v2需要改为IntWritable而不是nullwritable
        TwoInt K2 = new TwoInt();
        IntWritable v2= new IntWritable();
        @Override
        protected void map(LongWritable key, Text value,
                Mapper<LongWritable, Text, TwoInt, IntWritable>.Context context)
                throws IOException, InterruptedException {
            String[] splited = value.toString().split("	");
            K2.set(Integer.parseInt(splited[0]), Integer.parseInt(splited[1]));
            v2.set(Integer.parseInt(splited[1]));
            context.write(K2, v2);
        }
    }

    private static class MyReducer extends
            Reducer<TwoInt, IntWritable, IntWritable, IntWritable> {//k2,v2s,k3,v3
        IntWritable k3 = new IntWritable();
        IntWritable v3 = new IntWritable();
        @Override
        protected void reduce(
                TwoInt k2,
                Iterable<IntWritable> v2s,
                Reducer<TwoInt, IntWritable, IntWritable, IntWritable>.Context context)
                throws IOException, InterruptedException {
            int max=Integer.MIN_VALUE;
            for (IntWritable v2 : v2s) {
                if (v2.get()>max) {
                    max=v2.get();
                }
            }
            //每个组求得一个最大值可得到结果的序列
            v3.set(max);
            k3.set(k2.t1);//k2的第一列作为k3,因为k2为Twoint类型
            context.write(k3,v3);
        }
    }

    private static class TwoInt implements WritableComparable<TwoInt> {
        public int t1;
        public int t2;

        public void write(DataOutput out) throws IOException {
            out.writeInt(t1);
            out.writeInt(t2);
        }

        public void set(int t1, int t2) {
            this.t1 = t1;
            this.t2 = t2;
        }

        public void readFields(DataInput in) throws IOException {
            this.t1 = in.readInt();
            this.t2 = in.readInt();
        }

        public int compareTo(TwoInt o) {
            if (this.t1 == o.t1) { // 當第一列相等的時候，第二列升序排列
                return this.t2 - o.t2;
            }
            return this.t1 - o.t1;// 當第一列不相等的時候，按第一列升序排列
        }
        @Override
        public String toString() {
            return t1 + "	" + t2;
        }
    }
}

测试并运行结果如下：

[root@neusoft-master filecontent]# hadoop dfs -text  /out9/part-r-00000

 [root@neusoft-master filecontent]# hadoop dfs -text  /out10/part-r-00000

 

结果是正确无误的。

 END~