MapReduce实现每年的每月前两天温度最高排序

骑猪看日落 2022-05-20 08:40 170阅读 0赞

温度数据：

1949-10-01 14:21:02	34c
    1949-10-02 14:01:02	36c
    1950-01-01 11:21:02	32c
    1950-10-01 12:21:02	37c
    1951-12-01 11:21:02	23c
    1950-10-02 12:21:02	41c
    1950-10-03 11:21:02	27c
    1951-07-01 12:21:02	45c
    1951-07-02 11:21:02	46c
    1951-07-03 12:21:03	47c

需求：使用MapReduce实现每年的每月前两天最高的温度

思路：

1、mapper阶段：将数据中的年份按照升序排序，温度降序排序输出

2、reducer阶段：将mapper阶段处理好的年份数据的温度按照降序排序

自定义mapper

import org.apache.hadoop.io.IntWritable;
    import org.apache.hadoop.io.LongWritable;
    import org.apache.hadoop.io.Text;
    import org.apache.hadoop.mapreduce.Mapper;
    import org.apache.hadoop.util.StringUtils;
    
    import java.io.IOException;
    import java.text.ParseException;
    import java.text.SimpleDateFormat;
    import java.util.Calendar;
    
    //输出的key是什么，value是类型
    public class TQMapper extends Mapper<LongWritable,Text,Weather,IntWritable> {
        @Override
        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
           //读到这一行数据，转换成value这个对象，最后输出
            String[] strs = StringUtils.split(value.toString(), '\t');
    
            SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
            Calendar cal=Calendar.getInstance();
            try {
                cal.setTime(sdf.parse(strs[0]));
    
                Weather w = new Weather();
                w.setYear(cal.get(Calendar.YEAR));    //设置年份
                w.setMonth(cal.get(Calendar.MONTH)+1);//返回时日期从0开始，所以+1
                w.setDay(cal.get(Calendar.DAY_OF_MONTH));  //设置日期
    
                //1951-07-03 12:21:03	47c
                //设置温度，截取温度出来
                int wd= Integer.parseInt(strs[1].substring(0,strs[1].lastIndexOf("c")));
                w.setWd(wd);   //设置温度
    
                context.write(w, new IntWritable(wd));
    
            } catch (ParseException e) {
                e.printStackTrace();
            }
    
        }
    }

自定义reducer

import org.apache.hadoop.io.IntWritable;
    import org.apache.hadoop.io.NullWritable;
    import org.apache.hadoop.io.Text;
    import org.apache.hadoop.mapreduce.Reducer;
    
    import java.io.IOException;
    
    //输出自己封装一个语句输出，所以输出的value是NUllWritable
    
    public class TQReducer extends Reducer<Weather,IntWritable,Text,NullWritable> {
    
        @Override
        protected void reduce(Weather weather, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
            int flag=0;//控制前两条
    
            for (IntWritable value: values) {
                flag++;   
                if(flag>2){
                    break;
                }
                String msg=weather.getYear()+"-"+weather.getMonth()+"-"+weather.getDay()+"-"+value.get();
                context.write(new Text(msg),NullWritable.get());
            }
        }
    }

自定义排序Sort

import org.apache.hadoop.io.WritableComparable;
    import org.apache.hadoop.io.WritableComparator;
    
    //重写排序方法
    public class TQSort extends WritableComparator{
    
        //重写构造方法
        public TQSort() {
            //参数是：由于对象 是weather，所以是weather.class
            super(Weather.class,true);
        }
    
        //重写compare方法
    
        @Override
        public int compare(WritableComparable a, WritableComparable b) {
            Weather w1=(Weather)a;
            Weather w2=(Weather)b;
    
            //比较年月和温度，年份相同则为同一个对象再比较温度
            int c1=Integer.compare(w1.getYear(),w2.getYear());
                if(c1==0){
                    int c2=Integer.compare(w1.getMonth(),w2.getMonth());
    
                    if (c2==0){
                        //由于温度是降序排序,添加负号
                        return -Integer.compare(w1.getWd(),w2.getWd());
                    }
                    return c2;
                }
            return c1;
        }
    }

自定义分区partitioner

import org.apache.hadoop.io.IntWritable;
    import org.apache.hadoop.mapreduce.lib.partition.HashPartitioner;
    
    
    //输入是mapper阶段的输出
    public class TQPartition extends HashPartitioner<Weather,IntWritable> {
        @Override
        public int getPartition(Weather key, IntWritable value, int numReduceTasks) {
    
    
            //重写patition规则：1、满足业务需求  2、越简单越好
            return  (key.getYear()-1949)%numReduceTasks;
    
    
        }
    }

自定义分组Group

import org.apache.hadoop.io.WritableComparable;
    import org.apache.hadoop.io.WritableComparator;
    
    public class TQGroup extends WritableComparator {
    
        //重写构造方法
        public TQGroup() {
            //参数为：由于对象是weather，所以传入weather.class,同时传入true
            super(Weather.class,true);
        }
    
        //重写compare方法
        @Override
        public int compare(WritableComparable a, WritableComparable b) {
    
            //将相同年月的分成同一组
            Weather w1=(Weather)a;
            Weather w2=(Weather)b;
    
            //比较,分组返回的是相同的年份和月份
            int c1=Integer.compare(w1.getYear(),w2.getYear());
            if(c1==0){
                int c2=Integer.compare(w1.getMonth(),w2.getMonth());
                return c2;
            }
            return c1;
        }
    }

封装的WeatherBean

import org.apache.hadoop.io.WritableComparable;
    import java.io.DataInput;
    import java.io.DataOutput;
    import java.io.IOException;
    
    public class Weather implements WritableComparable<Weather>{
        private int year;
        private  int month;
        private int day;
    
        public int getYear() {
            return year;
        }
    
        public void setYear(int year) {
            this.year = year;
        }
    
        public int getMonth() {
            return month;
        }
    
        public void setMonth(int month) {
            this.month = month;
        }
    
        public int getDay() {
            return day;
        }
    
        public void setDay(int day) {
            this.day = day;
        }
    
        public int getWd() {
            return wd;
        }
    
        public void setWd(int wd) {
            this.wd = wd;
        }
    
        //温度
        private int wd;
    
    
    
    
        @Override
        public int compareTo(Weather w) {
            //compareto返回的时1，0，-1
            //先比较年在比较月，比较温度
            int c1=Integer.compare(this.year,w.getYear());
            //如果年相等,比较月
            if(c1==0){
                int c2=Integer.compare(this.month,w.getMonth());
                //如果月份相同再比较温度
                if (c2==0){
                    return Integer.compare(this.wd,w.getWd());
                }
                return c2;
            }
            return c1;
        }
    
        @Override
        public void write(DataOutput out) throws IOException {
            out.writeInt(year);
            out.writeInt(month);
            out.writeInt(day);
            out.writeInt(wd);
    
        }
    
        @Override
        public void readFields(DataInput in) throws IOException {
            this.year=in.readInt();
            this.month=in.readInt();
            this.day=in.readInt();
            this.wd=in.readInt();
        }
    }

Job

import com.doit.mr.flowprovince.FlowBean;
    import com.doit.mr.flowprovince.PhoneDriver;
    import com.test.MyGroup;
    import com.test.MyKey;
    import com.test.MyPartitioner;
    import com.test.MySort;
    import org.apache.hadoop.conf.Configuration;
    import org.apache.hadoop.fs.FileSystem;
    import org.apache.hadoop.fs.Path;
    import org.apache.hadoop.io.IntWritable;
    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 test.RunJob;
    
    public class TQJob {
        public static void main(String[] args) throws Exception {
            //创建Configuration对象
            Configuration configuration=new Configuration();
            Job job = Job.getInstance(configuration);
    
            //创建清理已存在的输出目录
            Path outputPath =new Path(args[1]);
            FileSystem fileSystem = FileSystem.get(configuration);
            if(fileSystem.exists(outputPath)){
                fileSystem.delete(outputPath,true);
                System.out.println("output file exit ,but it has deleted");
            }
    
    
            //设置job的处理类
            job.setJarByClass(TQJob.class);
    
            //设置作业处理的输入路径
            FileInputFormat.setInputPaths(job,new Path(args[0]));
    
            //设置map相关参数
            job.setMapperClass(TQMapper.class);
            job.setMapOutputKeyClass(Weather.class);
            job.setMapOutputValueClass(IntWritable.class);
    
            job.setPartitionerClass(TQPartition.class);
            job.setSortComparatorClass(TQSort.class);
            job.setGroupingComparatorClass(TQGroup.class);
            job.setNumReduceTasks(3);
    
            //设置reduce相关参数
            job.setReducerClass(TQReducer.class);
    
            //设置作业处理的输出路径
            FileOutputFormat.setOutputPath(job,new Path(args[1]));            //选lib
    
            //
            System.exit(job.waitForCompletion(true)?0:1);
        }
    }

运行结果：可以看到分成三个文件，其中第一个文件里面温度按照降序排序并且只取出前两天,格式为：年-月-日-温度

![70][]

[70]: /images/20220520/76f1c8bb96a249c2bfed0cf70ad5fc5d.png