Hadoop学习之路（十九）MapReduce框架排序

时间：2023-03-09 21:35:04
Hadoop学习之路（十九）MapReduce框架排序

流量统计项目案例

样本示例

Hadoop学习之路（十九）MapReduce框架排序

Hadoop学习之路（十九）MapReduce框架排序

需求

1、 统计每一个用户（手机号）所耗费的总上行流量、总下行流量，总流量

2、 得出上题结果的基础之上再加一个需求：将统计结果按照总流量倒序排序

3、 将流量汇总统计结果按照手机归属地不同省份输出到不同文件中

第一题

import java.io.IOException;

import org.apache.hadoop.conf.Configuration;

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;

/**

 *    第一题：统计每一个用户（手机号）所耗费的总上行流量、总下行流量，总流量

 */

public class FlowSumMR {

    public static void main(String[] args) throws Exception {

        Configuration conf = new Configuration();

        Job job = Job.getInstance(conf, "FlowSumMR");

        job.setJarByClass(FlowSumMR.class);

        job.setMapperClass(FlowSumMRMapper.class);

        job.setReducerClass(FlowSumMRReducer.class);

        job.setOutputKeyClass(Text.class);

        job.setOutputValueClass(Text.class);

        FileInputFormat.setInputPaths(job, new Path("E:/bigdata/flow/input/"));

        FileOutputFormat.setOutputPath(job, new Path("E:/bigdata/flow/output_sum"));

        boolean isDone = job.waitForCompletion(true);

        System.exit(isDone ? 0 : 1);

    }

    public static class FlowSumMRMapper extends Mapper<LongWritable, Text, Text, Text>{

        /**

         * value  =  1363157993044     18211575961    94-71-AC-CD-E6-18:CMCC-EASY    120.196.100.99

         * iface.qiyi.com    视频网站    15    12    1527    2106    200

         */

        @Override

        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {

            String[] split = value.toString().split("\t");

            String outkey = split[1];

            String outValue = split[8] + "\t" + split[9];

            context.write(new Text(outkey), new Text(outValue));

        }

    }

    public static class FlowSumMRReducer extends Reducer<Text, Text, Text, Text>{

        @Override

        protected void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException {

            int upFlow = 0;

            int downFlow = 0;

            int sumFlow = 0;

            for(Text t : values){

                String[] split = t.toString().split("\t");

                int upTempFlow = Integer.parseInt(split[0]);

                int downTempFlow = Integer.parseInt(split[1]);

                upFlow+=upTempFlow;

                downFlow +=  downTempFlow;

            }

            sumFlow = upFlow + downFlow;

            context.write(key, new Text(upFlow + "\t" + downFlow + "\t" + sumFlow));

        }

    }

}

第二题

import java.io.IOException;

import org.apache.hadoop.conf.Configuration;

import org.apache.hadoop.fs.Path;

import org.apache.hadoop.io.LongWritable;

import org.apache.hadoop.io.NullWritable;

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 comg.ghgj.mr.pojo.FlowBean;

/**

 * 需求： 第二个题目，就是对第一个题目的结果数据，进行按照总流量倒叙排序

 *

 *

 */

public class FlowSortMR {

    public static void main(String[] args) throws Exception {

        Configuration conf = new Configuration();

        Job job = Job.getInstance(conf, "FlowSumMR");

        job.setJarByClass(FlowSortMR.class);

        job.setMapperClass(FlowSortMRMapper.class);

        job.setReducerClass(FlowSortMRReducer.class);

        job.setOutputKeyClass(FlowBean.class);

        job.setOutputValueClass(NullWritable.class);

        FileInputFormat.setInputPaths(job, new Path("E:/bigdata/flow/output_sum"));

        FileOutputFormat.setOutputPath(job, new Path("E:/bigdata/flow/output_sort_777"));

        boolean isDone = job.waitForCompletion(true);

        System.exit(isDone ? 0 : 1);

    }

    public static class FlowSortMRMapper extends Mapper<LongWritable, Text, FlowBean, NullWritable>{

        /**

         * value  = 13602846565    26860680    40332600    67193280

         */

        @Override

        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {

            String[] split = value.toString().split("\t");

            FlowBean fb = new FlowBean(split[0], Long.parseLong(split[1]), Long.parseLong(split[2]));

            context.write(fb, NullWritable.get());

        }

    }

    public static class FlowSortMRReducer extends Reducer<FlowBean, NullWritable, FlowBean, NullWritable>{

        @Override

        protected void reduce(FlowBean key, Iterable<NullWritable> values, Context context)

                throws IOException, InterruptedException {

            for(NullWritable nvl : values){

                context.write(key, nvl);

            }

        }

    }

}

FlowBean.java

 import java.io.DataInput;

 import java.io.DataOutput;

 import java.io.IOException;

 import org.apache.hadoop.io.WritableComparable;

 /**

  * 第一，定义好属性

  * 第二，定义好属性的getter 和 setter方法

  * 第三，定义好构造方法（有参，无参）

  * 第四：定义好toString();

  *

  *

  * 详细解释：

  *

  * 如果一个自定义对象要作为key 必须要实现 WritableComparable 接口， 而不能实现 Writable, Comparable

  *

  * 如果一个自定义对象要作为value，那么只需要实现Writable接口即可

  */

 public class FlowBean implements WritableComparable<FlowBean>{

 //public class FlowBean implements Comparable<FlowBean>{

     private String phone;

     private long upFlow;

     private long downFlow;

     private long sumFlow;

     public String getPhone() {

         return phone;

     }

     public void setPhone(String phone) {

         this.phone = phone;

     }

     public long getUpFlow() {

         return upFlow;

     }

     public void setUpFlow(long upFlow) {

         this.upFlow = upFlow;

     }

     public long getDownFlow() {

         return downFlow;

     }

     public void setDownFlow(long downFlow) {

         this.downFlow = downFlow;

     }

     public long getSumFlow() {

         return sumFlow;

     }

     public void setSumFlow(long sumFlow) {

         this.sumFlow = sumFlow;

     }

     public FlowBean(String phone, long upFlow, long downFlow, long sumFlow) {

         super();

         this.phone = phone;

         this.upFlow = upFlow;

         this.downFlow = downFlow;

         this.sumFlow = sumFlow;

     }

     public FlowBean(String phone, long upFlow, long downFlow) {

         super();

         this.phone = phone;

         this.upFlow = upFlow;

         this.downFlow = downFlow;

         this.sumFlow = upFlow + downFlow;

     }

     public FlowBean() {

         super();

         // TODO Auto-generated constructor stub

     }

     @Override

     public String toString() {

         return  phone + "\t" + upFlow + "\t" + downFlow + "\t" + sumFlow;

     }

     /**

      * 把当前这个对象 --- 谁掉用这个write方法，谁就是当前对象

      *

      * FlowBean bean = new FlowBean();

      *

      * bean.write(out)    把bean这个对象的四个属性序列化出去

      *

      *  this = bean

      */

     @Override

     public void write(DataOutput out) throws IOException {

         // TODO Auto-generated method stub

         out.writeUTF(phone);

         out.writeLong(upFlow);

         out.writeLong(downFlow);

         out.writeLong(sumFlow);

     }

     //   序列化方法中的写出的字段顺序， 一定一定一定要和 反序列化中的 接收顺序一致。 类型也一定要一致

     /**

      * bean.readField();

      *

      *             upFlow =

      */

     @Override

     public void readFields(DataInput in) throws IOException {

         // TODO Auto-generated method stub

         phone = in.readUTF();

         upFlow = in.readLong();

         downFlow = in.readLong();

         sumFlow = in.readLong();

     }

     /**

      * Hadoop的序列化机制为什么不用   java自带的实现 Serializable这种方式？

      *

      * 本身Hadoop就是用来解决大数据问题的。

      *

      * 那么实现Serializable接口这种方式，在进行序列化的时候。除了会序列化属性值之外，还会携带很多跟当前这个对象的类相关的各种信息

      *

      * Hadoop采取了一种全新的序列化机制；只需要序列化 每个对象的属性值即可。

      */

     /*@Override

       public void readFields(DataInput in) throws IOException {

         value = in.readLong();

       }

       @Override

       public void write(DataOutput out) throws IOException {

         out.writeLong(value);

       }*/

     /**

      * 用来指定排序规则

      */

     @Override

     public int compareTo(FlowBean fb) {

         long diff = this.getSumFlow() - fb.getSumFlow();

         if(diff == 0){

             return 0;

         }else{

             return diff > 0 ? -1 : 1;

         }

     }

 }

第三题

package comg.ghgj.mr.flow;

import java.io.IOException;

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.mapreduce.lib.partition.ProvincePartitioner;

public class FlowPartitionerMR {

    public static void main(String[] args) throws Exception {

        Configuration conf = new Configuration();

        FileSystem fs = FileSystem.get(conf);

        Job job = Job.getInstance(conf, "FlowSumMR");

        job.setJarByClass(FlowPartitionerMR.class);

        job.setMapperClass(FlowPartitionerMRMapper.class);

        job.setReducerClass(FlowPartitionerMRReducer.class);

        job.setOutputKeyClass(Text.class);

        job.setOutputValueClass(Text.class);

        /**

         * 非常重要的两句代码

         */

        job.setPartitionerClass(ProvincePartitioner.class);

        job.setNumReduceTasks(10);

        FileInputFormat.setInputPaths(job, new Path("E:\\bigdata\\flow\\input"));

        Path outputPath = new Path("E:\\bigdata\\flow\\output_ptn2");

        if(fs.exists(outputPath)){

            fs.delete(outputPath, true);

        }

        FileOutputFormat.setOutputPath(job, outputPath);

        boolean isDone = job.waitForCompletion(true);

        System.exit(isDone ? 0 : 1);

    }

    public static class FlowPartitionerMRMapper extends Mapper<LongWritable, Text, Text, Text>{

        /**

         * value  =  13502468823    101663100    1529437140    1631100240

         */

        @Override

        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {

            String[] split = value.toString().split("\t");

            String outkey = split[1];

            String outValue = split[8] + "\t" + split[9];

            context.write(new Text(outkey), new Text(outValue));

        }

    }

    public static class FlowPartitionerMRReducer extends Reducer<Text, Text, Text, Text>{

        @Override

        protected void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException {

            int upFlow = 0;

            int downFlow = 0;

            int sumFlow = 0;

            for(Text t : values){

                String[] split = t.toString().split("\t");

                int upTempFlow = Integer.parseInt(split[0]);

                int downTempFlow = Integer.parseInt(split[1]);

                upFlow+=upTempFlow;

                downFlow +=  downTempFlow;

            }

            sumFlow = upFlow + downFlow;

            context.write(key, new Text(upFlow + "\t" + downFlow + "\t" + sumFlow));

        }

    }

}

相关文章