MapReduce案例-流量统计

一、MapReduce案例-流量统计

源数据

源代码

1：需求一: 统计求和

统计每个手机号的上行数据包总和，下行数据包总和，上行总流量之和，下行总流量之和分析：以手机号码作为key值，上行流量，下行流量，上行总流量，下行总流量四个字段作为value值，然后以这个key，和value作为map阶段的输出，reduce阶段的输入

1.1: 自定义map的输出value对象FlowBean

package flowcount;

import org.apache.hadoop.io.Writable;

import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;


public class FlowBean implements Writable {
    private Integer upFlow;  //上行数据包数
    private Integer downFlow;  //下行数据包数
    private Integer upCountFlow; //上行流量总和
    private Integer 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 upFlow +
                "t" + downFlow +
                "t" + upCountFlow +
                "t" + 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();
    }
}

1.2：定义FlowMapper类

package flowcount;

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.yarn.webapp.hamlet.Hamlet;

import java.io.IOException;


public class FlowCountMapper extends Mapper {
 

    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        //1:拆分行文本数据,得到手机号--->K2
        String[] split = value.toString().split("t");
        String phoneNum = split[1];
        //2:创建FlowBean对象,并从行文本数据拆分出流量的四个四段,并将四个流量字段的值赋给FlowBean对象
        FlowBean flowBean = new FlowBean();

        flowBean.setUpFlow(Integer.parseInt(split[6]));
        flowBean.setDownFlow(Integer.parseInt(split[7]));
        flowBean.setUpCountFlow(Integer.parseInt(split[8]));
        flowBean.setDownCountFlow(Integer.parseInt(split[9]));

        //3:将K2和V2写入上下文中
        context.write(new Text(phoneNum), flowBean);
    }
}

1.3：定义FlowReducer类

package flowcount;

import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;


public class FlowCountReducer extends Reducer {
    @Override
    protected void reduce(Text key, Iterable values, Context context) throws IOException, InterruptedException {
        //1:遍历集合,并将集合中的对应的四个字段累计
        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();
        }
        //2:创建FlowBean对象,并给对象赋值  V3
        FlowBean flowBean = new FlowBean();
        flowBean.setUpFlow(upFlow);
        flowBean.setDownFlow(downFlow);
        flowBean.setUpCountFlow(upCountFlow);
        flowBean.setDownCountFlow(downCountFlow);

        //3:将K3和V3下入上下文中
        context.write(key, flowBean);
    }
}

1.4：程序main函数入口FlowMain

package flowcount;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;



public class jobMain extends Configured implements Tool {
    //该方法用于指定一个job任务
    @Override
    public int run(String[] args) throws Exception {
        //1、创建一个job对象
        Job job = Job.getInstance(super.getConf(), "mapreduce_flowcount");
        //2、配置job任务的8个对象
        //第一步 指定文件的读取方式和路径
        job.setInputFormatClass(TextInputFormat.class);
        
        TextInputFormat.addInputPath(job,new Path("file:///D:\input\flowcount"));
        //第二步 指定map阶段的处理方式和数据类型
        job.setMapperClass(FlowCountMapper.class);
        //设置map阶段k2的类型
        job.setMapOutputKeyClass(Text.class);
        //设置map阶段v2的类型
        job.setMapOutputValueClass(FlowBean.class);
        //第三(分区) 四(排序) 第五步(规约) 六(分组) 步 采用默认方式
        //第七步 指定reduce阶段的处理方式和数据类型
        job.setReducerClass(FlowCountReducer.class);
        //设置k3  v3类型
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(FlowBean.class);
        //第八步 设置输出类型
        job.setOutputFormatClass(TextOutputFormat.class);
        //本地运行模式
        TextOutputFormat.setOutputPath(job,new Path("file:///D:\output\flowcount"));
        //等待任务结束
        boolean b = job.waitForCompletion(true);
        return b ? 0:1;
    }

    public static void main(String[] args) throws Exception {
        Configuration configuration=new Configuration();
        //启动job任务
        int run = ToolRunner.run(configuration, new jobMain(), args);
        System.exit(run);
    }
}

1.5：运行截图

2：需求二: 上行流量倒序排序(递减排序)

分析，以需求一的输出数据作为排序的输入数据，自定义FlowBean,以FlowBean为map输出的key，以手机号作为Map输出的value，因为MapReduce程序会对Map阶段输出的key进行排序

2.1：定义FlowBean实现WritableComparable实现比较排序

Java 的 compareTo 方法说明:

compareTo 方法用于将当前对象与方法的参数进行比较。
如果指定的数与参数相等返回 0。
如果指定的数小于参数返回 -1。
如果指定的数大于参数返回 1。

例如：o1.compareTo(o2); 返回正数的话，当前对象(调用 compareTo 方法的对象 o1)要排在比较对象(compareTo 传参对象 o2)后面，返回负数的话，放在前面

package flowcount.sort;

import org.apache.hadoop.io.Writable;
import org.apache.hadoop.io.WritableComparable;

import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;


public class FlowBean implements WritableComparable {
    private Integer upFlow;  //上行数据包数
    private Integer downFlow;  //下行数据包数
    private Integer upCountFlow; //上行流量总和
    private Integer 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 upFlow +
                "t" + downFlow +
                "t" + upCountFlow +
                "t" + 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();
    }

    //指定排序规则
    @Override
    public int compareTo(FlowBean flowBean) {
        // return this.upFlow.compareTo(flowBean.getUpFlow()) * -1;
        return  flowBean.upFlow - this.upFlow ;
    }
}

2.2：定义FlowMapper

package flowcount.sort;

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import java.io.IOException;


public class FlowSortMapper extends Mapper {
    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        //1:拆分行文本数据(V1),得到四个流量字段,并封装FlowBean对象---->K2
        String[] split = value.toString().split("t");

        FlowBean flowBean = new FlowBean();

        flowBean.setUpFlow(Integer.parseInt(split[1]));
        flowBean.setDownFlow(Integer.parseInt(split[2]));
        flowBean.setUpCountFlow(Integer.parseInt(split[3]));
        flowBean.setDownCountFlow(Integer.parseInt(split[4]));

        //2:通过行文本数据,得到手机号--->V2
        String phoneNum = split[0];

        //3:将K2和V2下入上下文中
        context.write(flowBean, new Text(phoneNum));
    }
}

2.3:定义FlowReducer

package flowcount.sort;

import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;


public class FlowSortReducer extends Reducer {
    @Override
    protected void reduce(FlowBean key, Iterable values, Context context) throws IOException, InterruptedException {
        //1:遍历集合,取出 K3,并将K3和V3写入上下文中
        for (Text value : values) {
            context.write(value, key);
        }

    }
}

2.4:程序main函数入口

package flowcount.sort;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;
import sort.SortBean;



public class jobMain extends Configured implements Tool {
    //该方法用于指定一个job任务
    @Override
    public int run(String[] args) throws Exception {
        //1、创建一个job对象
        Job job = Job.getInstance(super.getConf(), "mapreduce_flowcount_cort");
        //2、配置job任务的8个对象
        //第一步 指定文件的读取方式和路径
        job.setInputFormatClass(TextInputFormat.class);
        
        TextInputFormat.addInputPath(job,new Path("file:///D:\output\flowcount"));
        //第二步 指定map阶段的处理方式和数据类型
        job.setMapperClass(FlowSortMapper.class);
        //设置map阶段k2的类型
        job.setMapOutputKeyClass(FlowBean.class);
        //设置map阶段v2的类型
        job.setMapOutputValueClass(Text.class);
        //第三(分区) 四(排序) 第五步(规约) 六(分组) 步 采用默认方式
        //第七步 指定reduce阶段的处理方式和数据类型
        job.setReducerClass(FlowSortReducer.class);
        //设置k3  v3类型
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(SortBean.class);
        //第八步 设置输出类型
        job.setOutputFormatClass(TextOutputFormat.class);
        //本地运行模式
        TextOutputFormat.setOutputPath(job,new Path("file:///D:\output\flowcount_sort"));
        //等待任务结束
        boolean b = job.waitForCompletion(true);
        return b ? 0:1;
    }

    public static void main(String[] args) throws Exception {
        Configuration configuration=new Configuration();
        //启动job任务
        int run = ToolRunner.run(configuration, new jobMain(), args);
        System.exit(run);
    }
}

2.5：运行截图

3：需求三: 手机号码分区

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

135 开头数据到一个分区文件

136 开头数据到一个分区文件

137 开头数据到一个分区文件

其他分区

3.1：自定义分区

package flowcount.partition;

import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Partitioner;


public class FlowCountPartition extends Partitioner {
    
    @Override
    public int getPartition(Text text, FlowBean flowBean, int i) {
        //1:获取手机号
        String phoneNum = text.toString();

        //2:判断手机号以什么开头,返回对应的分区编号(0-3)
        if(phoneNum.startsWith("135")){
            return  0;
        }else  if(phoneNum.startsWith("136")){
            return  1;
        }else  if(phoneNum.startsWith("137")){
            return  2;
        }else{
            return  3;
        }
    }
}

3.2：作业运行设置

job.setPartitionerClass(FlowPartition.class);
job.setNumReduceTasks(4);

3.3：运行结果

MapReduce案例-流量统计

大数据系统相关栏目本月热门文章