15.大数据---Mapreduce案例之---统计手机号耗费的总上行流量、下行流量、总流量_学无止境的大象_上行流量

irpas 02-07 5218

Mapreduce案例之—统计手机号耗费的总上行流量、下行流量、总流量

1.需求：统计每一个手机号耗费的总上行流量、下行流量、总流量

2.数据准备： 2.1 输入数据格式：时间戳、电话号码、基站的物理地址、访问网址的ip、网站域名、数据包、接包数、上行/传流量、下行/载流量、响应码这些就是10个字段的数据；我们可以通过自己去模拟数据； 2.2 最终输出的数据格式：手机号码上行流量下行流量总流量

3.基本思路： 3.1 Map阶段：（1）读取一行数据，转换为字符串类型

（2）切分字段

（3）抽取手机号、上行流量、下行流量

（4）以手机号为key，bean对象（上行流量、下行流量、总流量）为value 进行封装

（5）文件写出，即context.write(手机号,bean)

3.2 Reduce阶段（1）遍历集合上行流量和下行流量总和得到总流量

（2）实现自定义的bean来封装流量信息，并将bean作为map输出的key来传输

（3）MR程序在处理数据的过程中会对数据排序(map输出的kv对传输到reduce之前，会排序)，排序的依据是map输出的key

4.代码实现 4.1 编写流量统计的bean对象–FlowBean.java package com.dataflow;

import org.apache.hadoop.io.Writable;

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

//1实现writable方法 public class FlowBean implements Writable {

private long upflow; private long downflow; private long sumflow; //必须要有空参构造，为了以后反射用 public FlowBean() { super(); } public FlowBean(long upflow, long downflow) { super(); this.upflow = upflow; this.downflow = downflow; this.sumflow = upflow+downflow; } public void set(long upflow, long downflow) { this.upflow = upflow; this.downflow = downflow; this.sumflow = upflow+downflow; } //序列化的方法 ---- 对数据进行读和写的具体的操作； public void write(DataOutput out) throws IOException { out.writeLong(upflow); out.writeLong(downflow); out.writeLong(sumflow); //反序列化方法 //注意序列化方法和反序列化方法顺序必须保持一致 } public void readFields(DataInput in) throws IOException { this.upflow=in.readLong(); this.downflow=in.readLong(); this.sumflow=in.readLong(); } @Override public String toString() { return upflow + "\t" + downflow + "\t" + sumflow; } public void setUpflow(long upflow) { this.upflow = upflow; } public long getUpflow() { return 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; }

}

4.2 Mapper阶段–FlowBeanMapper.java package com.dataflow;

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

import java.io.IOException;

public class FlowMappper extends Mapper<LongWritable, Text, Text, FlowBean> { Text k = new Text(); // 对象的方式接数据 FlowBean v = new FlowBean();

@Override protected void map(LongWritable key, Text value, Context context)throws IOException, InterruptedException { String line = value.toString(); String[] fields = line.split("\t"); String phNum = fields[1]; long upFlow = Long.parseLong(fields[fields.length - 3]); long downFlow = Long.parseLong(fields[fields.length - 2]);

// 以对象的方式把数据接收 k.set(phNum); v.set(upFlow, downFlow); context.write(k, v); } }

4.3 Reduce阶段–FlowBeanReducer.java package com.dataflow;

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

import java.io.IOException;

public class FlowReducer extends Reducer<Text, FlowBean, Text, FlowBean> {

@Override protected void reduce(Text key, Iterable<FlowBean> values, Context context) throws IOException, InterruptedException { long sumUpFlow = 0; long sumDownFlow = 0; System.out.println(values); for (FlowBean flowBean : values) { sumUpFlow += flowBean.getUpflow(); sumDownFlow += flowBean.getDownflow(); } FlowBean v = new FlowBean(sumUpFlow, sumDownFlow); context.write(key, v); }

}

4.4 Driver 阶段–FlowBeanDriver.java—启动程序 package com.dataflow;

import org.apache.hadoop.conf.Configuration; 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.FileInputFormat; import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import java.io.IOException;

public class FlowDriver { public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException { Configuration Configuration=new Configuration(); Job job= Job.getInstance(Configuration);

job.setJarByClass(FlowDriver.class); job.setMapperClass(FlowMappper.class); job.setReducerClass(FlowReducer.class); job.setMapOutputKeyClass(Text.class); job.setMapOutputValueClass(FlowBean.class); job.setOutputKeyClass(Text.class); job.setOutputValueClass(FlowBean.class); FileInputFormat.setInputPaths(job, new Path("E:/dataflow.txt")); FileOutputFormat.setOutputPath(job, new Path("E:/BigData"));

// FileInputFormat.setInputPaths(job, new Path(args[0])); // FileOutputFormat.setOutputPath(job, new Path(args[1]));

boolean result=job.waitForCompletion(true); System.out.println(result?"老铁,没毛病。就算出来的结果了！！！！！！！！":"哥们,出BUG了,赶快去修改一下！！！"); }

}

5.运行结果