一、需求说明

1、数据文件说明

hdfs中有一些存储温度的数据文件,以文本形式存储,示例如下:

日期和时间中间是空格,为整体,表示检测站点监测的时间,后面是检测的温度,中间通过制表符 t 相隔。

「 Hadoop」mapreduce对温度数据进行自定义排序、分组、分区等 【转载】

2、需求

  1. 计算在1949-1955年中,每年的温度降序排序且每年单独一个文件输出存储

需要进行自定义分区、自定义分组、自定义排序。

二、解决

1、思路

  1. 按照年份升序排序再按照每年的温度降序排序
  2. 按照年份进行分组,每一年份对应一个reduce task

2、自定义mapper输出类型keypair

可以看出,每一行温度姑且称为一个数据,每个数据中有两部分,一部分是时间,另一部分是温度。

因此map输出必须使用自定义的格式输出,并且输出之后需要自定义进行排序和分组等操作,默认的那些都不管用了。

定义keypair

自定义的输出类型因为要将map的输出放到reduce中去运行,因此需要实现hadoop的writablecomparable的接口,并且该接口的模板变量也得是keypair,就像是longwritable一个意思(查看longwritable的定义就可以知道)

实现writablecomparable 的接口,就必须重写write/readfileds/compareto三个方法,依次作用于序列化/反序列化/比较

同时需要重写tostring和hashcode避免equals的问题。

keypair定义如下

值得注意的是:在进行序列化输出的时候也就是write,里面用了将标准格式的时间(文件中显示的格式时间)进行的时间的转换,用了datainput和dataoutput

import org.apache.hadoop.io.writablecomparable;
import java.io.datainput;
import java.io.dataoutput;
import java.io.ioexception;

/**
 * project   : hadooptest2
 * package   : com.mapreducetest.temp
 * user      : postbird @ http://www.ptbird.cn
 * time      : 2017-01-19 21:53
 */

/**
 * 为温度和年份封装成对象
 * year表示年份 而temp为温度
 */
public class keypair implements writablecomparable<keypair>{
    //年份
    private int year;
    //温度
    private int temp;

    public void setyear(int year) {
        this.year = year;
    }

    public void settemp(int temp) {
        this.temp = temp;
    }

    public int getyear() {
        return year;
    }

    public int gettemp() {
        return temp;
    }
    @override
    public int compareto(keypair o) {
        //传过来的对象和当前的year比较 相等为0 不相等为1
        int result=integer.compare(year,o.getyear());
        if(result != 0){
            //两个year不相等
            return 0;
        }
        //如果年份相等 比较温度
        return integer.compare(temp,o.gettemp());
    }

    @override
    //序列化
    public void write(dataoutput dataoutput) throws ioexception {
       dataoutput.writeint(year);
       dataoutput.writeint(temp);
    }

    @override
    //反序列化
    public void readfields(datainput datainput) throws ioexception {
        this.year=datainput.readint();
        this.temp=datainput.readint();
    }

    @override
    public string tostring() {
        return year+"\t"+temp;
    }

    @override
    public int hashcode() {
        return new integer(year+temp).hashcode();
    }
}

3、自定义分组

将同一年监测的温度放到一起,因此需要对年份进行比较。

因此比较输入的数据中的年份即可,注意此时比较的都是keypair的类型,map出来的输出也是这个类型。

因为继承了writablecomparator,因此需要重写compare方法,比较的是keypair(keypair实现了writablecomparable接口),实际比较的使他们的年份,年份相同则得到0

/**
 * project   : hadooptest2
 * package   : com.mapreducetest.temp
 * user      : postbird @ http://www.ptbird.cn
 * time      : 2017-01-19 22:08
 */

import org.apache.hadoop.io.writablecomparable;
import org.apache.hadoop.io.writablecomparator;

/**
 *  为温度分组 比较年份即可
 */
public class grouptemp extends writablecomparator{

    public grouptemp() {
        super(keypair.class,true);
    }
    @override
    public int compare(writablecomparable a, writablecomparable b) {
        //年份相同返回的是0
        keypair o1=(keypair)a;
        keypair o2=(keypair)b;
        return integer.compare(o1.getyear(),o2.getyear());
    }
}

4、自定义分区

自定义分区的目的是在根据年份分好了组之后,将不同的年份创建不同的reduce task任务,因此需要对年份处理。

import org.apache.hadoop.io.text;
import org.apache.hadoop.mapreduce.partitioner;

/**
 * project   : hadooptest2
 * package   : com.mapreducetest.temp
 * user      : postbird @ http://www.ptbird.cn
 * time      : 2017-01-19 22:17
 */

//自定义分区
//每一个年份生成一个reduce任务
public class firstpartition extends partitioner<keypair,text>{
    @override
    public int getpartition(keypair key, text value, int num) {
        //按照年份进行分区 年份相同,返回的是同一个值
        return (key.getyear()*127)%num;
    }
}

5、自定义排序

最终还是比较的是温度的排序,因此这部分也是非常重要的。

根据上面的需求,需要对年份进行生序排序,而对温度进行降序排序,首选比较条件是年份.

/**
 * project   : hadooptest2
 * package   : com.mapreducetest.temp
 * user      : postbird @ http://www.ptbird.cn
 * time      : 2017-01-19 22:08
 */

import org.apache.hadoop.io.writablecomparable;
import org.apache.hadoop.io.writablecomparator;

/**
 *  为温度排序的封装类
 */
public class sorttemp extends writablecomparator{

    public sorttemp() {
        super(keypair.class,true);
    }
    //自定义排序
    @override
    public int compare(writablecomparable a, writablecomparable b) {
        //按照年份升序排序 按照温度降序排序
        keypair o1=(keypair)a;
        keypair o2=(keypair)b;
        int result=integer.compare(o1.getyear(),o2.getyear());
        //比较年份 如果年份不相等
        if(result != 0){
            return result;
        }
        //两个年份相等 对温度进行降序排序,注意 - 号
        return -integer.compare(o1.gettemp(),o2.gettemp());
    }
}

6、mapreduce程序的编写

几个值得注意的点:

    1. 数据文件中前面的时间是字符串,但是我们的keypair的set却不是字符串,因此需要进行字符串转日期的format操作,使用的是simpledateformat,格式自然是"yyyy-mm-dd hh:mm:ss"了。
    2. 输入每行数据之后,通过正则匹配"t"的制表符,然后将温度和时间分开,将时间format并得到年份,将第二部分字符串去掉“℃”的符号得到数字,然后创建keypair类型的数据,在输出即可。
  1. 每个年份都生成一个reduce task依据就是自定义分区中对年份进行了比较处理,为了简单就把map的输出结果在reduce中再输出一次,三个reduce task,就会生成三个输出文件。
  2. 因为使用了自定义的排序,分组,分区,因此就需要进行指定相关的class,同时也需要执行reduce task的数量。
  3. 其实最后客户端还是八股文的固定形式而已,只不过多了自定义的指定,没有别的。
import org.apache.hadoop.conf.configuration;
import org.apache.hadoop.fs.filesystem;
import org.apache.hadoop.fs.path;
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;
import org.apache.hadoop.util.genericoptionsparser;

import java.io.ioexception;
import java.net.uri;
import java.text.simpledateformat;
import java.util.calendar;
import java.util.date;

/**
 * project   : hadooptest2
 * package   : com.mapreducetest.temp
 * user      : postbird @ http://www.ptbird.cn
 * time      : 2017-01-19 22:28
 */
public class runtempjob {
    //字符串转日期format
    public static simpledateformat sdf=new simpledateformat("yyyy-mm-dd hh:mm:ss");
    /**
     * mapper
     * 输出的key是自定义的keypair
     */
    static class tempmapper extends mapper<longwritable,text,keypair,text>{
        protected void map(longwritable key,text value,context context) throws ioexception,interruptedexception{
            string line=value.tostring();
            //1949-10-01 14:21:02    34℃
            // 前面是空格 时间和温度通过\t分割
            string[] ss=line.split("\t");
//            system.err.println(ss.length);
            if(ss.length==2){
                try{
                    //获得日期
                    date date=sdf.parse(ss[0]);
                    calendar c=calendar.getinstance();
                    c.settime(date);
                    int year=c.get(1);//得到年份
                    //字符串截取得到温度,去掉℃
                    string temp = ss[1].substring(0,ss[1].indexof("℃"));
                    //创建输出key 类型为keypair
                    keypair kp=new keypair();
                    kp.setyear(year);
                    kp.settemp(integer.parseint(temp));
                    //输出
                    context.write(kp,value);
                }catch(exception ex){
                    ex.printstacktrace();
                }
            }
        }
    }
    /**
     *  reduce 区域
     *  map的输出是reduce的输出
     */
    static class tempreducer extends reducer<keypair,text,keypair,text> {
        @override
        protected void reduce(keypair kp, iterable<text> values, context context) throws ioexception, interruptedexception {
            for (text value:values){
                context.write(kp,value);
            }
        }
    }

    //client
    public static void main(string args[]) throws ioexception, interruptedexception{
        //获取配置
        configuration conf=new configuration();

        //修改命令行的配置
        string[] otherargs = new genericoptionsparser(conf, args).getremainingargs();
        if (otherargs.length != 2) {
            system.err.println("usage: temp <in> <out>");
            system.exit(2);
        }
        //创建job
        job job=new job(conf,"temp");
        //1.设置job运行的类
        job.setjarbyclass(runtempjob.class);
        //2.设置map和reduce的类
        job.setmapperclass(runtempjob.tempmapper.class);
        job.setreducerclass(runtempjob.tempreducer.class);
        //3.设置map的输出的key和value 的类型
        job.setmapoutputkeyclass(keypair.class);
        job.setmapoutputvalueclass(text.class);
        //4.设置输入文件的目录和输出文件的目录
        fileinputformat.addinputpath(job,new path(otherargs[0]));
        fileoutputformat.setoutputpath(job,new path(otherargs[1]));
        //5.设置reduce task的数量 每个年份对应一个reduce task
        job.setnumreducetasks(3);//3个年份
        //5.设置partition sort group的class
        job.setpartitionerclass(firstpartition.class);
        job.setsortcomparatorclass(sorttemp.class);
        job.setgroupingcomparatorclass(grouptemp.class);
        //6.提交job 等待运行结束并在客户端显示运行信息
        boolean issuccess= false;
        try {
            issuccess = job.waitforcompletion(true);
        } catch (classnotfoundexception e) {
            e.printstacktrace();
        }
        //7.结束程序
        system.exit(issuccess ?0:1);
    }
}

三、生成效果:

hdfs中三个reduce task会生成三个输出。

「 Hadoop」mapreduce对温度数据进行自定义排序、分组、分区等 【转载】

每个输出文件都是每年中的温度的排序结果:

「 Hadoop」mapreduce对温度数据进行自定义排序、分组、分区等 【转载】

可以看出,1951是map(也可以说是keypair)输出的年份,46是温度,而后面是将text又输出了一次,每一年都是根据需求降序排序的。)