hadoop-ykt(自定义key)

1.自定义key对象，实现WritableComparable接口

package com.zhangdan.ykt;

import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
import org.apache.hadoop.io.WritableComparable;

/**
 * 自定义的key类型
 * 
 * @author vlab
 *
 */
public class GdzBeans implements WritableComparable<GdzBeans> {
    private String key1;
    private String key2;

    public GdzBeans() {// 自定义数据类型必须有一个无参的构造方法，为了防止反射

    }

    public GdzBeans(String key1, String key2) {
        this.key1 = key1;
        this.key2 = key2;
    }

    public String getKey1() {
        return key1;
    }

    public String getKey2() {
        return key2;
    }

    public void setKey(String key1, String key2) {
        this.key1 = key1;
        this.key2 = key2;
    }

    public void setKey1(String key1) {
        this.key1 = key1;
    }

    public void setKey2(String key2) {
        this.key2 = key2;
    }

    @Override
    public void readFields(DataInput in) throws IOException {
        this.key1 = in.readUTF();
        this.key2 = in.readUTF();

    }

    @Override
    public void write(DataOutput out) throws IOException {
        out.writeUTF(key1);
        out.writeUTF(key2);
    }

    @Override
    public int compareTo(GdzBeans o) {
        int cmp1 = key1.compareTo(o.key1);
        if (cmp1 != 0) {
            return cmp1;
        }
        int cmp2 = key2.compareTo(o.key2);
        return cmp2;
    }

    @Override
    public boolean equals(Object obj) {
        if (obj instanceof GdzBeans) {
            GdzBeans pw = (GdzBeans) obj;
            return key1.equals(pw.key1) && key2.equals(pw.key2);
        }
        return false;
    }

    @Override
    public int hashCode() {
        return this.key1.hashCode() * 3 + this.key2.hashCode() * 5;
    }

    @Override
    public String toString() {
        return this.key1 + "," + this.key2;
    }

}

 自定义的key要实现WritableComparable，然后复写对应的几个方法，
Writable---toString()、equals()、hasCode()
Comparable---compareTo(),这个是关键，在这里可以对key排序，之前总不明白这个是什么个逻辑，除了很多错误，比较坑.

2.mapreduce

package com.zhangdan.ykt;

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.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;

public class GetConnectionCount {
    private static final String Input_Path = "hdfs://192.168.1.104:9000/user/vlab/yktdata/pid07_result04/connection/part-r-*";
    private static final String Output_Path = "hdfs://192.168.1.104:9000/user/vlab/yktdata/pid07_result04/connectioncount";
    // private static final String Input_Path =
    // "hdfs://192.168.1.104:9000/user/vlab/yktdata/pid06_result09/connection/part-r-*";
    // private static final String Output_Path =
    // "hdfs://192.168.1.104:9000/user/vlab/yktdata/pid06_result09test/connectioncount";

    public static class MyMap extends Mapper<LongWritable, Text, GdzBeans, IntWritable> {
        IntWritable one = new IntWritable(1);

        public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
            String line = value.toString();
            // System.out.println(line);
            String[] tt;
            tt = line.split(",");
            context.write(new GdzBeans(tt[0], tt[1]), one);
        }
    }

    public static class MyReduce extends Reducer<GdzBeans, IntWritable, GdzBeans, IntWritable> {
        public void reduce(GdzBeans key, Iterable<IntWritable> values, Context context)
                throws IOException, InterruptedException {
            int n = 0;
            for (IntWritable i : values) {
                n += i.get();
            }
            if (key.getKey1() != null && key.getKey2() != null && key.getKey1().trim() != ""
                    && key.getKey2().trim() != "")
                context.write(key, new IntWritable(n));
        }
    }

    public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
        Configuration conf = new Configuration();
        conf.set("mapred.textoutputformat.ignoreseparator", "true"); // 去掉key/value对的自动补tab符
        conf.set("mapred.textoutputformat.separator", ",");// 设置以“，”为key/value的分隔符
        Job job = new Job(conf);
        job.setJarByClass(GdzTest.class);
        job.setJobName("GetConnectionCount");

        job.setMapperClass(MyMap.class);
        job.setMapOutputKeyClass(GdzBeans.class);
        job.setMapOutputValueClass(IntWritable.class);

        job.setReducerClass(MyReduce.class);
        job.setCombinerClass(MyReduce.class);
        job.setOutputKeyClass(GdzBeans.class);
        job.setOutputValueClass(IntWritable.class);

        job.setInputFormatClass(TextInputFormat.class);
        job.setOutputFormatClass(TextOutputFormat.class);

        FileInputFormat.addInputPath(job, new Path(Input_Path));
        FileOutputFormat.setOutputPath(job, new Path(Output_Path));

        System.exit(job.waitForCompletion(true) ? 0 : 1);

    }
}

 这个这里有很多改进，很多之前不理解的也慢慢理解了，但是像这个我想实现二次排序还是没有得到比较满意的解决，
看了好多大家写的二次排序的代码，但是总是和我这情况有点不一致，复杂的类我还得慢慢适应。