Hadoop（九）：Shuffle组件

package com.rzp.pojo;
?
import org.apache.hadoop.io.WritableComparable;
?
import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
//作为key，要实现WritableComparable
public class CustomerValue implements WritableComparable<CustomerValue> {
    private String id;
    private long time;
?
    public int compareTo(CustomerValue o) {
        int tmp = this.id.compareTo(o.id);
        if (tmp!=0){return tmp;}
        tmp = Long.compare(this.time,o.time);
        return tmp;
    }
?
    public void write(DataOutput out) throws IOException {
        out.writeUTF(this.id);
        out.writeLong(this.time);
    }
?
    public void readFields(DataInput in) throws IOException {
        this.id = in.readUTF();
        this.time = in.readLong();
    }

package com.rzp.pojo;
?
        import org.apache.hadoop.mapreduce.Partitioner;
        import org.apache.hadoop.mapreduce.lib.partition.HashPartitioner;
?
//自定义分区类
public class CustomerPartitioner extends Partitioner<CustomerValue,Object> {
    private Partitioner<String,Object> proxy = new HashPartitioner<String,Object> ();
    
    //numPartitions一般Hadoop会直接输入ReducerTask的数量
    @Override
    public int getPartition(CustomerValue key, Object o, int numPartitions) {
        return this.proxy.getPartition(key.getId(),o,numPartitions);
    }
}
?

  /** Use {@link Object#hashCode()} to partition. */
  public int getPartition(K key, V value,
                          int numReduceTasks) {
    return (key.hashCode() & Integer.MAX_VALUE) % numReduceTasks;
  }

package com.rzp.utils;
?
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.log4j.Logger;
?
import java.io.IOException;
?
//输入输出的键值对一摸一样
public class WordCountCombiner extends Reducer<Text, LongWritable,Text, LongWritable> {
    private static final Logger logger = Logger.getLogger(WordCountCombiner.class);
?
    private LongWritable count = new LongWritable();
?
    @Override
    protected void setup(Context context) throws IOException, InterruptedException {
        super.setup(context);
        logger.debug("调用combiner");
    }
?
    @Override
    protected void reduce(Text key, Iterable<LongWritable> values, Context context) throws IOException, InterruptedException {
        logger.debug("执行combiner"+key);
        
        //先进行一部分的合并累加操作
        int sum = 0;
        for (LongWritable value :values){
            sum += value.get();
        }
        this.count.set(sum);
        context.write(key,this.count);
    }
?
    @Override
    protected void cleanup(Context context) throws IOException, InterruptedException {
        super.cleanup(context);
        logger.debug("完成combiner的调用");
    }
}

      //设置combainer
        job.setCombinerClass(WordCountCombiner.class);
 

package com.rzp.pojo;
?
import org.apache.hadoop.io.WritableComparable;
import org.apache.hadoop.io.WritableComparator;
?
public class CustomGrouping extends WritableComparator {
    public CustomGrouping(){
        //调用父类构造器，对父类要使用的key值赋值
        //第一个参数是该grouping进行分组对应的key类
        //第二个参数写true，表示直接创建对象
        super(CustomerValue.class,true);
    }
?
    @Override
    public int compare(WritableComparable a, WritableComparable b) {
        CustomerValue cv1 = (CustomerValue)a;
        CustomerValue cv2 = (CustomerValue)b;
?
        return cv1.getId().compareTo(cv2.getId());
    }
}

package com.rzp.sorttwice;
?
import org.apache.hadoop.io.WritableComparable;
?
import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
?
//
public class SortOutputKey implements WritableComparable<SortOutputKey> {
    private String uid;
    private String time;
    public int compareTo(SortOutputKey o) {
        //主要用于排序，先把需要group的字段排序，然后再进行其他字段排序
        //拿this比较o，那么得到结果是从小到大
        int tmp = this.uid.compareTo(o.uid);
        if (tmp!=0){
            return tmp;
        }
        //tmp = this.time.compareTo(o.time); //time从小到大
        tmp = o.time.compareTo(this.time);
        return tmp;
    }
?
    public void write(DataOutput out) throws IOException {
        out.writeUTF(this.uid);
        out.writeUTF(this.time);
    }
?
    public void readFields(DataInput in) throws IOException {
        this.uid = in.readUTF();
        this.time = in.readUTF();
    }
?
    public String getUid() {
        return uid;
    }
?
    public void setUid(String uid) {
        this.uid = uid;
    }
?
    public String getTime() {
        return time;
    }
?
    public void setTime(String time) {
        this.time = time;
    }
?
    @Override
    public boolean equals(Object o) {
        if (this == o) return true;
        if (o == null || getClass() != o.getClass()) return false;
?
        SortOutputKey that = (SortOutputKey) o;
?
        if (uid != null ? !uid.equals(that.uid) : that.uid != null) return false;
        return time != null ? time.equals(that.time) : that.time == null;
    }
?
    @Override
    public int hashCode() {
        int result = uid != null ? uid.hashCode() : 0;
        result = 31 * result + (time != null ? time.hashCode() : 0);
        return result;
    }
?
    @Override
    public String toString() {
        return "SortOutputKey{" +
                "uid=‘" + uid + ‘\‘‘ +
                ", time=‘" + time + ‘\‘‘ +
                ‘}‘;
    }
}

package com.rzp.sorttwice;
?
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.WritableComparable;
import org.apache.hadoop.mapreduce.Mapper;
?
import java.io.IOException;
?
public class SortMapper extends Mapper<Object, Text,SortOutputKey,Text> {
    private SortOutputKey outputKey = new SortOutputKey();
    private Text outPutValue = new Text();
    @Override
    protected void map(Object key, Text value, Context context) throws IOException, InterruptedException {
        String line = value.toString();
        String[] splits = line.split(",");
        if (splits!=null&&splits.length==3){
            String uid = splits[0];
            String time = splits[1];
            String url = splits[2];
?
            if (uid!=null&&time!=null && url!=null){
                this.outputKey.setUid(uid);
                this.outputKey.setTime(time);
                this.outPutValue.set(url);
                context.write(this.outputKey,this.outPutValue);
            }
        }
    }
}

package com.rzp.sorttwice;
?
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;
?
import java.io.IOException;
?
?
//在二次排序的实现中，进入reduce方法的key/value键值对是随时更新的，会将同一组key一起进行处理
public class SortReducer extends Reducer<SortOutputKey, Text,Text,Text> {
    private Text outputKey = new Text();
    private Text outputvalue = new Text();
    private int i = 0;
    @Override
    protected void reduce(SortOutputKey key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
        StringBuffer sb = new StringBuffer();
        for (Text value : values) {
            System.out.println(i+":"+key);
            sb.append(value.toString()).append(",");
        }
        this.outputKey.set(key.getUid());
        this.outputvalue.set(sb.substring(0,sb.length()-1));
        i++;
        context.write(outputKey,outputvalue);
    }
}

package com.rzp.sorttwice;
?
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;
?
public class SortRunner implements Tool {
    private Configuration conf = null;
    public int run(String[] args) throws Exception {
        Configuration conf = this.getConf();
        Job job = Job.getInstance(conf,"sorting");
?
        job.setJarByClass(SortRunner.class);
        job.setMapperClass(SortMapper.class);
        job.setMapOutputKeyClass(SortOutputKey.class);
        job.setMapOutputValueClass(Text.class);
        FileInputFormat.setInputPaths(job,new Path("D:\\Hoptest\\sort\\in\\data.txt"));
?
        job.setReducerClass(SortReducer.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(Text.class);
        FileOutputFormat.setOutputPath(job,new Path("D:\\Hoptest\\sort\\out"));
        return job.waitForCompletion(true)?0:-1;
?
    }
?
    public void setConf(Configuration conf) {
        conf.set("mapreduce.framework.name","local");
        this.conf=conf;
    }
?
    public Configuration getConf() {
        return this.conf;
    }
?
    public static void main(String[] args) throws Exception {
        ToolRunner.run(new SortRunner(),args);
    }
?
}

package com.rzp.sorttwice;
?
import org.apache.hadoop.io.WritableComparable;
import org.apache.hadoop.io.WritableComparator;
?
public class SortGrouping extends WritableComparator {
    public SortGrouping(){
        super(SortOutputKey.class,true);
    }
?
    @Override
    public int compare(WritableComparable a, WritableComparable b) {
        SortOutputKey k1 = (SortOutputKey) a;
        SortOutputKey k2 = (SortOutputKey) b;
        return k1.getUid().compareTo(k2.getUid());
    }
}
?

package com.rzp.sorttwice;
?
import org.apache.hadoop.mapreduce.Partitioner;
import org.apache.hadoop.mapreduce.lib.partition.HashPartitioner;
?
public class SortPartitioner extends Partitioner<SortOutputKey,Object> {
    private Partitioner<Object,Object> proxy = new HashPartitioner<Object, Object>();
?
    @Override
    public int getPartition(SortOutputKey key, Object value, int numPartitions) {
        return this.proxy.getPartition(key.getUid(),value,numPartitions);
    }
}
?

        //一般在二次排序中，分区和分组的字段要一样
        //partitioner
        job.setPartitionerClass(SortPartitioner.class);
        //grouping
        job.setGroupingComparatorClass(SortGrouping.class);