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

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.运行结果

15.大数据---Mapreduce案例之---统计手机号耗费的总上行流量、下行流量、总流量相关推荐

流量汇总案例（一）之统计手机号耗费的总上行流量、下行流量、总流量（序列化）
1)需求: 统计每一个手机号耗费的总上行流量.下行流量.总流量 2)数据准备 1363157985066 13726230503 00-FD-07-A4-72-B8:CMCC 120.196.100. ...
统计手机号耗费的总上行流量、下行流量、总流量（序列化）
自定义排序 1.需求: 2.数据准备: 3.分析: 4.代码实现: (1)FlowBean类: (2)MapWritable类: (3)ReduceWritable类: (4)MainWritable ...
2017最新总结政府大数据应用案例及启示
在我国,政府部门掌握着全社会量最大.最核心的数据.以往地方政府提振经济一般是招房地产.工厂等,随着土地及人口红利殆尽,大数据成为与水电煤等一样重要的生产资料,成为继土地之后政府最重要的资产.了解政府大 ...
“毒舌”专家解析大数据应用案例Part1—高德交通大数据应用
秦雯:[友盟+]CMO,1995年进入互联网行业,先后就职于瀛海威.千龙新闻网和新浪网.2007年曾创立缔元信.网络数据公司,为各大媒体网站及电商网站.广告主和代理机构.政府主管部门提供数据平台解决方 ...
全面落地，2014大数据应用案例TOP100
在当前的互联网领域,大数据的应用已十分广泛,尤其以企业为主,企业成为大数据应用的主体.大数据真能改变企业的运作方式吗?答案毋庸置疑是肯定的.随着企业开始利用大数据,我们每天都会看到大数据新的奇妙的应用 ...
寻找最具创新的大数据应用案例，下一个就是你！
由中国互联网协会主办,[友盟+]再次独家承办"2016中国互联网大会全域大数据应用论坛"6.21在北京国际中心一层第三会议室召开! 下面就为大家揭开第一个谜团: [友盟+]联合阿里 ...
“毒舌”专家解析大数据应用案例Part3—大数据打造互联网+物流菜鸟物流大数据应用分析...
从大数据概念到大数据应用,各种公开演讲活动中,秦雯女士都在给大数据泼"冷水",因此,行业里都称她大数据的"毒舌"代表. [友盟+]CMO 秦雯 1995年进入互 ...
“毒舌”专家解析大数据应用案例Part2—三星盖乐世社区 DSP跨屏投放
从大数据概念到大数据应用,各种公开演讲活动中,秦雯女士都在给大数据泼"冷水",因此,行业里都称她大数据的"毒舌"代表. [友盟+]CMO 秦雯 1995年进入互 ...
我们整理了2017年最新政府大数据应用案例！
本文长度为6000字,建议阅读10分钟本文为你介绍政府拥有数据资产类型和地方政府大数据应用案例及应用启示. 在我国,政府部门掌握着全社会量最大.最核心的数据.以往地方政府提振经济一般是招房地产.工厂 ...

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

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

15.大数据---Mapreduce案例之---统计手机号耗费的总上行流量、下行流量、总流量相关推荐

最新文章

热门文章