MapReduce习题练习 08

我不是女神ヾ 2022-03-02 15:42 177阅读 0赞

# 1. 统计求和 #

## 1.1 需求 ##

统计每个手机号的上行流量总和，下行流量总和，上行总流量之和，下行总流量之和

## 1.2 分析 ##

以手机号码作为key值，上行流量，下行流量，上行总流量，下行总流量四个字段作为value值，然后以这个key，和value作为map阶段的输出，reduce阶段的输入  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MjY0MTkwOQ_size_16_color_FFFFFF_t_70]

## 1.3 代码 ##

### 1.3.1 自定义map的输出value对象FlowBean ###

public class FlowBean implements Writable { 
        private Integer upFlow;
        private Integer  downFlow;
        private Integer upCountFlow;
        private Integer downCountFlow;
        @Override
        public void write(DataOutput out) throws IOException { 
            out.writeInt(upFlow);
            out.writeInt(downFlow);
            out.writeInt(upCountFlow);
            out.writeInt(downCountFlow);
        }
        @Override
        public void readFields(DataInput in) throws IOException { 
            this.upFlow = in.readInt();
            this.downFlow = in.readInt();
            this.upCountFlow = in.readInt();
            this.downCountFlow = in.readInt();
        }
        public FlowBean() { 
        }
        public FlowBean(Integer upFlow, Integer downFlow, Integer upCountFlow, Integer downCountFlow) { 
            this.upFlow = upFlow;
            this.downFlow = downFlow;
            this.upCountFlow = upCountFlow;
            this.downCountFlow = downCountFlow;
        }
        public Integer getUpFlow() { 
            return upFlow;
        }
        public void setUpFlow(Integer upFlow) { 
            this.upFlow = upFlow;
        }
        public Integer getDownFlow() { 
            return downFlow;
        }
        public void setDownFlow(Integer downFlow) { 
            this.downFlow = downFlow;
        }
        public Integer getUpCountFlow() { 
            return upCountFlow;
        }
        public void setUpCountFlow(Integer upCountFlow) { 
            this.upCountFlow = upCountFlow;
        }
        public Integer getDownCountFlow() { 
            return downCountFlow;
        }
        public void setDownCountFlow(Integer downCountFlow) { 
            this.downCountFlow = downCountFlow;
        }
        @Override
        public String toString() { 
            return "FlowBean{" +
                    "upFlow=" + upFlow +
                    ", downFlow=" + downFlow +
                    ", upCountFlow=" + upCountFlow +
                    ", downCountFlow=" + downCountFlow +
                    '}';
        }
    }

### 1.3.2 定义FlowMapper类 ###

public class FlowMapper extends Mapper<LongWritable,Text,Text,FlowBean> { 
        FlowBean flowBean = new FlowBean();
        @Override
        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException { 
            String s = value.toString();
            String[] split = s.split("\t");
            flowBean.setUpFlow(Integer.parseInt(split[6]));
            flowBean.setDownFlow(Integer.parseInt(split[7]));
            flowBean.setUpCountFlow(Integer.parseInt(split[8]));
            flowBean.setDownCountFlow(Integer.parseInt(split[9]));
            context.write(new Text(split[1]),flowBean);
        }
    }

### 1.3.3 定义FlowReducer类 ###

public class FlowReducer extends Reducer<Text,FlowBean,Text,FlowBean> { 
        private FlowBean flowBean = new FlowBean();
        @Override
        protected void reduce(Text key, Iterable<FlowBean> values, Context context) throws IOException, InterruptedException { 
             Integer upFlow = 0;
             Integer  downFlow = 0;
             Integer upCountFlow = 0;
             Integer downCountFlow = 0;
            for (FlowBean value : values) { 
                upFlow += value.getUpFlow();
                downFlow += value.getDownFlow();
                upCountFlow += value.getUpCountFlow();
                downCountFlow += value.getDownCountFlow();
            }
            flowBean.setUpFlow(upFlow);
            flowBean.setDownFlow(downFlow);
            flowBean.setUpCountFlow(upCountFlow);
            flowBean.setDownCountFlow(downCountFlow);
            context.write(key,flowBean);
        }
    }

### 1.3.4 程序main函数入口FlowMain ###

public class FlowBeanMain extends Configured implements Tool { 
        @Override
        public int run(String[] args) throws Exception { 
            Job job = Job.getInstance(super.getConf(), FlowBeanMain.class.getSimpleName());
            job.setJarByClass(FlowBeanMain.class);
            job.setInputFormatClass(TextInputFormat.class);
            TextInputFormat.addInputPath(job,new Path("file:///F:\\传智播客大数据离线阶段课程资料\\4、大数据离线第四天\\流量统计\\input"));
            job.setMapperClass(FlowMapper.class);
            job.setMapOutputKeyClass(Text.class);
            job.setMapOutputValueClass(FlowBean.class);
            job.setReducerClass(FlowReducer.class);
            job.setOutputKeyClass(Text.class);
            job.setOutputValueClass(FlowBean.class);
            job.setOutputFormatClass(TextOutputFormat.class);
            TextOutputFormat.setOutputPath(job,new Path("file:///F:\\传智播客大数据离线阶段课程资料\\4、大数据离线第四天\\流量统计\\out"));
    
            boolean b = job.waitForCompletion(true);
            return b?0:1;
        }
        public static void main(String[] args) throws Exception { 
            ToolRunner.run(new Configuration(),new FlowBeanMain(),args);
        }
    }

# 2. 上行流量倒序排序（递减排序） #

## 2.1 分析 ##

分析，以需求一的输出数据作为排序的输入数据，自定义FlowBean,以FlowBean为map输出的key，以手机号作为Map输出的value，因为MapReduce程序会对Map阶段输出的key进行排序  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MjY0MTkwOQ_size_16_color_FFFFFF_t_70 1]

## 2.2 代码 ##

### 2.2.1 定义FlowBean实现WritableComparable实现比较排序 ###

java 的compareTo方法说明  
compareTo 方法用于将当前对象与方法的参数进行比较。  
如果指定的数与参数相等返回 0。  
如果指定的数小于参数返回 -1。  
如果指定的数大于参数返回 1。  
例如：o1.compareTo(o2);  
返回正数的话，当前对象（调用 compareTo 方法的对象 o1）要排在比较对象（compareTo 传参对象 o2）后面，返回负数的话，放在前面。

public class FlowBean implements WritableComparable<FlowBean> { 
        private Integer upFlow;
        private Integer  downFlow;
        private Integer upCountFlow;
        private Integer downCountFlow;
        public FlowBean() { 
        }
        public FlowBean(Integer upFlow, Integer downFlow, Integer upCountFlow, Integer downCountFlow) { 
            this.upFlow = upFlow;
            this.downFlow = downFlow;
            this.upCountFlow = upCountFlow;
            this.downCountFlow = downCountFlow;
        }
        @Override
        public void write(DataOutput out) throws IOException { 
            out.writeInt(upFlow);
            out.writeInt(downFlow);
            out.writeInt(upCountFlow);
            out.writeInt(downCountFlow);
        }
        @Override
        public void readFields(DataInput in) throws IOException { 
            upFlow = in.readInt();
            downFlow = in.readInt();
            upCountFlow = in.readInt();
            downCountFlow = in.readInt();
        }
        public Integer getUpFlow() { 
            return upFlow;
        }
        public void setUpFlow(Integer upFlow) { 
            this.upFlow = upFlow;
        }
        public Integer getDownFlow() { 
            return downFlow;
        }
        public void setDownFlow(Integer downFlow) { 
            this.downFlow = downFlow;
        }
        public Integer getUpCountFlow() { 
            return upCountFlow;
        }
        public void setUpCountFlow(Integer upCountFlow) { 
            this.upCountFlow = upCountFlow;
        }
        public Integer getDownCountFlow() { 
            return downCountFlow;
        }
        public void setDownCountFlow(Integer downCountFlow) { 
            this.downCountFlow = downCountFlow;
        }
        @Override
        public String toString() { 
            return upFlow+"\t"+downFlow+"\t"+upCountFlow+"\t"+downCountFlow;
        }
        @Override
        public int compareTo(FlowBean o) { 
            return this.upCountFlow > o.upCountFlow ?-1:1;
        }
    }

### 2.2.2 定义FlowMapper ###

public class FlowMapper extends Mapper<LongWritable,Text,FlowBean,Text> { 
         Text outKey = new Text();
         FlowBean flowBean = new FlowBean();
        @Override
        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException { 
            String[] split = value.toString().split("\t");
            flowBean.setUpFlow(Integer.parseInt(split[1]));
            flowBean.setDownFlow(Integer.parseInt(split[2]));
            flowBean.setUpCountFlow(Integer.parseInt(split[3]));
            flowBean.setDownCountFlow(Integer.parseInt(split[4]));
            outKey.set(split[0]);
            context.write(flowBean,outKey);
        }
    }

### 2.2.3 定义FlowReducer ###

public class FlowReducer extends Reducer<FlowBean,Text,Text,FlowBean> { 
        FlowBean flowBean = new FlowBean();
        @Override
        protected void reduce(FlowBean key, Iterable<Text> values, Context context) throws IOException, InterruptedException { 
           context.write(values.iterator().next(),key);
        }
    }

### 2.2.4 程序main函数入口 ###

public class FlowMain extends Configured implements Tool { 
        @Override
        public int run(String[] args) throws Exception { 
            Configuration conf = super.getConf();
            conf.set("mapreduce.framework.name","local");
            Job job = Job.getInstance(conf, FlowMain.class.getSimpleName());
            job.setJarByClass(FlowMain.class);
            job.setInputFormatClass(TextInputFormat.class);
            TextInputFormat.addInputPath(job,new Path("file:///L:\\大数据离线阶段备课教案以及资料文档——by老王\\4、大数据离线第四天\\流量统计\\output"));
            job.setMapperClass(FlowMapper.class);
            job.setMapOutputKeyClass(FlowBean.class);
            job.setMapOutputValueClass(Text.class);
            job.setReducerClass(FlowReducer.class);
            job.setOutputKeyClass(Text.class);
            job.setOutputValueClass(FlowBean.class);
            TextOutputFormat.setOutputPath(job,new Path("file:///L:\\大数据离线阶段备课教案以及资料文档——by老王\\4、大数据离线第四天\\流量统计\\outputsort"));
            job.setOutputFormatClass(TextOutputFormat.class);
            boolean b = job.waitForCompletion(true);
            return b?0:1;
        }
        public static void main(String[] args) throws Exception { 
            Configuration configuration = new Configuration();
            int run = ToolRunner.run(configuration, new FlowMain(), args);
            System.exit(run);
        }
    }

# 3. 手机号码分区 #

## 3.1 需求 ##

在需求一的基础上，继续完善，将不同的手机号分到不同的数据文件的当中去，需要自定义分区来实现，这里我们自定义来模拟分区，将以下数字开头的手机号进行分开

137 开头数据到一个分区文件
    138 开头数据到一个分区文件
    139 开头数据到一个分区文件
    135 开头数据到一个分区文件
    136 开头数据到一个分区文件
    其他分区

## 3.2 代码 ##

### 3.2.1 自定义分区 ###

public class FlowPartition extends Partitioner<Text,FlowBean> { 
        @Override
        public int getPartition(Text text, FlowBean flowBean, int i) { 
            String line = text.toString();
            if (line.startsWith("135")){ 
                return 0;
            }else if(line.startsWith("136")){ 
                return 1;
            }else if(line.startsWith("137")){ 
                return 2;
            }else if(line.startsWith("138")){ 
                return 3;
            }else if(line.startsWith("139")){ 
                return 4;
            }else{ 
                return 5;
            }
        }
    }

### 3.2.2 作业运行添加分区设置 ###

job.setPartitionerClass(FlowPartition.class);

### 3.2.3 更改输入与输出路径，并打包到集群上面去运行 ###

TextInputFormat.addInputPath(job,new Path("hdfs://node01:8020/partition_flow/"));
    TextOutputFormat.setOutputPath(job,new Path("hdfs://node01:8020/partition_out"));

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MjY0MTkwOQ_size_16_color_FFFFFF_t_70]: /images/20220302/3236c9d8a6da461cbdffd47b2447ad0d.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MjY0MTkwOQ_size_16_color_FFFFFF_t_70 1]: /images/20220302/7d29180d450d4a638a7955e7f9b1aaa0.png