HadoopMapReduce实战案例：手机流量使用统计分析

本文旨在通过一个实际的案例，帮助读者理解如何使用Hadoop MapReduce来解决大数据处理中的具体问题。我们将通过统计手机用户的流量使用情况，来展示MapReduce在数据处理方面的强大功能。

待统计的流量数据样例：

以下是部分需要统计的流量数据示例：

1363157985066 13726230503 00-FD-07-A4-72-B8:CMCC 120.196.100.82 i02.c.aliimg.com 24 27 2481 24681 200
1363157995052 13826544101 5C-0E-8B-C7-F1-E0:CMCC 120.197.40.4 4 0 264 0 200
1363157991076 13926435656 20-10-7A-28-CC-0A:CMCC 120.196.100.99 2 4 132 1512 200
...

每条记录包含多个字段，其中关键字段包括手机号码、上行流量、下行流量等。我们的目标是统计每个手机号码的上行流量、下行流量及总流量。

为了实现这一目标，我们设计了相应的MapReduce程序，主要包括以下几个部分：

• FlowBean类：用于封装流量数据，实现了Writable接口以便于数据的序列化与反序列化。
• PhoneFlowMapper类：作为Map阶段的处理逻辑，负责解析输入数据并输出中间结果。
• PhoneFlowReducer类：作为Reduce阶段的处理逻辑，负责对中间结果进行聚合计算，最终输出每个手机号的总流量信息。
• PhoneFlowApp类：作为程序的入口，配置MapReduce任务的各项参数并提交执行。

FlowBean类代码示例：

public class FlowBean implements Writable {
    private long upFlow;
    private long downFlow;
    private long sumFlow;

    public FlowBean() {}

    public FlowBean(long upFlow, long downFlow, long sumFlow) {
        this.upFlow = upFlow;
        this.downFlow = downFlow;
        this.sumFlow = sumFlow;
    }

    public void setFlowData(long upFlow, long downFlow) {
        this.upFlow = upFlow;
        this.downFlow = downFlow;
        this.sumFlow = upFlow + downFlow;
    }

    // Getters and setters

    @Override
    public void write(DataOutput out) throws IOException {
        out.writeLong(this.upFlow);
        out.writeLong(this.downFlow);
        out.writeLong(this.sumFlow);
    }

    @Override
    public void readFields(DataInput in) throws IOException {
        this.upFlow = in.readLong();
        this.downFlow = in.readLong();
        this.sumFlow = in.readLong();
    }
}

PhoneFlowMapper类代码示例：

public class PhoneFlowMapper extends Mapper {
    private FlowBean flowBean = new FlowBean();
    private Text keyText = new Text();

    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        String line = value.toString();
        String[] fields = line.split("\t");
        String phOneNum= fields[1];
        long upflow = Long.parseLong(fields[fields.length - 3]);
        long downflow = Long.parseLong(fields[fields.length - 2]);
        flowBean.setFlowData(upflow, downflow);
        keyText.set(phoneNum);
        context.write(keyText, flowBean);
    }
}

PhoneFlowReducer类代码示例：

public class PhoneFlowReducer extends Reducer {
    private FlowBean flowBean = new FlowBean();

    @Override
    protected void reduce(Text key, Iterable values, Context context) throws IOException, InterruptedException {
        long sumDownFlow = 0;
        long sumUpFlow = 0;
        for (FlowBean value : values) {
            sumUpFlow += value.getUpFlow();
            sumDownFlow += value.getDownFlow();
        }
        flowBean.setFlowData(sumUpFlow, sumDownFlow);
        context.write(key, flowBean);
    }
}

PhoneFlowApp类代码示例：

public class PhoneFlowApp {
    public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
        Configuration cOnf= new Configuration();
        Job job = Job.getInstance(conf);
        job.setJarByClass(PhoneFlowApp.class);
        job.setMapperClass(PhoneFlowMapper.class);
        job.setReducerClass(PhoneFlowReducer.class);
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(FlowBean.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(FlowBean.class);
        FileInputFormat.setInputPaths(job, new Path("hdfs://hadoop-001:9000/flowcount/input/HTTP_20130313143750.dat"));
        FileOutputFormat.setOutputPath(job, new Path("hdfs://hadoop-001:9000/flowcount/output/"));
        boolean res = job.waitForCompletion(true);
        System.exit(res ? 0 : 1);
    }
}

通过上述代码，我们可以有效地统计出每个手机号的上行流量、下行流量及总流量，并将结果输出。此外，如果需要根据总流量进行排序，并将结果输出到不同的文件中，可以通过实现自定义的Partitioner类来实现这一功能。

自定义Partitioner类代码示例：

public class FlowPartitioner extends Partitioner {
    @Override
    public int getPartition(FlowBean flowBean, Text text, int numPartitions) {
        String phOneNum= text.toString();
        String headThreePhOneNum= phoneNum.substring(0, 3);
        if (headThreePhoneNum.equals("134")) return 0;
        else if (headThreePhoneNum.equals("135")) return 1;
        else if (headThreePhoneNum.equals("136")) return 2;
        else if (headThreePhoneNum.equals("137")) return 3;
        else if (headThreePhoneNum.equals("138")) return 4;
        else return 5;
    }
}

通过设置自定义的Partitioner类，我们可以根据手机号前缀将数据分配到不同的Reducer中，从而实现数据的分类输出。