概述
一、partitioner类
Partitioner 的功能是在 Map 端对 key 进行分区。Map端最终处理的<key,value>对需要发送到 Reduce 端去合并，合并的时候，相同分区的<key,value>对会被分配到同一个 Reduce 上，这个分配过程就是由 Partitioner（分区）决定的。
MapReduce 默认的Partitioner 是HashPartitioner。其计算方法如下：
- Partitioner 先计算 key 的散列值（通常是 MD5 值）。
- 通过 Reduce 个数执行取模运算：Key.hashCode%numReduce。
二、按号段统计手机号码
1、题目描述。
按号段统计手机号码，手机号前三位相同的统计数据单独放在一个结果中。统计方式按135字段、136字段、137字段、138字段、139字段及其其它号段分区，共6个分区。表信息如下：
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
1363154400022 13926251106 5C-0E-8B-8B-B1-50:CMCC 120.197.40.4 4 0 240 0 200
1363157993044 18211575961 94-71-AC-CD-E6-18:CMCC-EASY 120.196.100.99 iface.qiyi.com 视频网站 15 12 1527 2106 200
1363157995074 84138413 5C-0E-8B-8C-E8-20:7DaysInn 120.197.40.4 122.72.52.12 20 16 4116 1432 200
1363157993055 13560439658 C4-17-FE-BA-DE-D9:CMCC 120.196.100.99 18 15 1116 954 200
1363157995033 15920133257 5C-0E-8B-C7-BA-20:CMCC 120.197.40.4 sug.so.360.cn 信息安全 20 20 3156 2936 200
1363157983019 13719199419 68-A1-B7-03-07-B1:CMCC-EASY 120.196.100.82 4 0 240 0 200
1363157984041 13660577991 5C-0E-8B-92-5C-20:CMCC-EASY 120.197.40.4 s19.cnzz.com 站点统计 24 9 6960 690 200
1363157973098 15013685858 5C-0E-8B-C7-F7-90:CMCC 120.197.40.4 rank.ie.sogou.com 搜索引擎 28 27 3659 3538 200
1363157986029 15989002119 E8-99-C4-4E-93-E0:CMCC-EASY 120.196.100.99 www.umeng.com 站点统计 3 3 1938 180 200
1363157992093 13560439658 C4-17-FE-BA-DE-D9:CMCC 120.196.100.99 15 9 918 4938 200
1363157986041 13480253104 5C-0E-8B-C7-FC-80:CMCC-EASY 120.197.40.4 3 3 180 180 200
1363157984040 13602846565 5C-0E-8B-8B-B6-00:CMCC 120.197.40.4 2052.flash2-http.qq.com 综合门户 15 12 1938 2910 200
1363157995093 13922314466 00-FD-07-A2-EC-BA:CMCC 120.196.100.82 img.qfc.cn 12 12 3008 3720 200
1363157982040 13502468823 5C-0A-5B-6A-0B-D4:CMCC-EASY 120.196.100.99 y0.ifengimg.com 综合门户 57 102 7335 110349 200
1363157986072 18320173382 84-25-DB-4F-10-1A:CMCC-EASY 120.196.100.99 input.shouji.sogou.com 搜索引擎 21 18 9531 2412 200
1363157990043 13925057413 00-1F-64-E1-E6-9A:CMCC 120.196.100.55 t3.baidu.com 搜索引擎 69 63 11058 48243 200
1363157988072 13760778710 00-FD-07-A4-7B-08:CMCC 120.196.100.82 2 2 120 120 200
1363157985066 13560436666 00-FD-07-A4-72-B8:CMCC 120.196.100.82 i02.c.aliimg.com 24 27 2481 24681 200
1363157993055 13560436666 C4-17-FE-BA-DE-D9:CMCC 120.196.100.99 18 15 1116 954 200
2、关键代码：
1）Mobile 实体类
package cn.kgc.mr.partitioner;
import org.apache.hadoop.io.Writable;
import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
public class FlowBean implements Writable {
private long upFlow;
private long downFlow;
private long sumFlow;
/*
序列化
*/
@Override
public void write(DataOutput dataOutput) throws IOException {
dataOutput.writeLong(upFlow);
dataOutput.writeLong(downFlow);
dataOutput.writeLong(sumFlow);
}
/*
反序列化
注意：序列化和反序列化字段的顺序需要保持一致
*/
@Override
public void readFields(DataInput dataInput) throws IOException {
this.upFlow = dataInput.readLong();
this.downFlow = dataInput.readLong();
this.sumFlow = dataInput.readLong();
}
public FlowBean(){
}
public FlowBean(long upFlow, long downFlow, long sumFlow) {
this.upFlow = upFlow;
this.downFlow = downFlow;
this.sumFlow = sumFlow;
}
//自己创建一个set方法
public void set(long upFlow ,long downFlow){
this.upFlow = upFlow;
this.downFlow = downFlow;
this.sumFlow = upFlow+downFlow;
}
@Override
public String toString() {
return "FlowBean{" +
"upFlow=" + upFlow +
", downFlow=" + downFlow +
", sumFlow=" + sumFlow +
'}';
}
public long getUpFlow() {
return upFlow;
}
public void setUpFlow(long upFlow) {
this.upFlow = 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;
}
}
2）自定义分区
package cn.kgc.mr.partitioner;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Partitioner;
public class ProvincePartitioner extends Partitioner<Text, FlowBean> {
@Override
public int getPartition(Text key, FlowBean value, int i) {
String perNum = key.toString().substring(0,3);
int partition = 4;
if("136".equals(perNum)){
partition=0;
}else if("137".equals(perNum)){
partition=1;
}else if("138".equals(perNum)){
partition=2;
}else if("139".equals(perNum)){
partition=3;
}
return partition;
}
}
3）MapReduce 功能
//Mapper类
package cn.kgc.mr.partitioner;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
import java.io.IOException;
public class FlowMapper 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 {
//1、将文本转换成string
String line = value.toString();
//2、将字符串切割
String[] fields = line.split("\s+");
//3、执行我们的业务逻辑
String phoneNumber = fields[1];
//取出上行和下行流量
long upFlow = Long.parseLong(fields[fields.length-3]) ;
long dowmFlow = Long.parseLong(fields[fields.length-2]) ;
k.set(phoneNumber);
v.set(upFlow,dowmFlow);
context.write(k,v);
}
}
//Reduce类
package cn.kgc.mr.partitioner;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;
import java.io.IOException;
public class FlowReduce extends Reducer<Text, FlowBean,Text, FlowBean> {
FlowBean v = new FlowBean();
@Override
protected void reduce(Text key, Iterable<FlowBean> values, Context context) throws IOException, InterruptedException {
//reduce的输入大概是这样的 ("13560439658", (FlowBean(918,4938),FlowBean(116,954)))
//创建两个初始值，用于累加操作
long sum_upFlow = 0;
long sum_downFlow = 0;
//执行累加操作
for (FlowBean flowBean : values) {
sum_upFlow += flowBean.getUpFlow();
sum_downFlow += flowBean.getDownFlow();
}
//将结果写出
v.set(sum_upFlow,sum_downFlow);
context.write(key,v);
}
}
//Main 驱动类
package cn.kgc.mr.partitioner;
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 {
//1、创建配置文件
Configuration conf = new Configuration();
Job job = Job.getInstance(conf,"flowCount");
//2、设置jar的位置
job.setJarByClass(FlowDriver.class);
//3、设置map和reduce的位置
job.setMapperClass(FlowMapper.class);
job.setReducerClass(FlowReduce.class);
//设置分区位置
job.setPartitionerClass(ProvincePartitioner.class);
//设置分区数量，大于，多余的分区会有空白文件
//1个全部输出在一个文件夹
//小于5大于1会报错
job.setNumReduceTasks(5);//5个或1个，
//4、设置map输出的key，value类型
job.setMapOutputKeyClass(Text.class);
job.setMapOutputValueClass(FlowBean.class);
//5、设置reduce输出的key，value类型
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(FlowBean.class);
//6、设置输出的路径
FileInputFormat.setInputPaths(job, new Path("file:///D:\Idea\ideaMaven\hadoopdfs1\data\fcinput"));
FileOutputFormat.setOutputPath(job, new Path("file:///D:\Idea\ideaMaven\hadoopdfs1\data\partitionerOutput"));
//7、提交程序运行
boolean result = job.waitForCompletion(true);
System.out.println(result ? 0:1);
}
}
最后
以上就是凶狠白猫为你收集整理的MapReduce实例 - partitioner 分区实现按号段统计手机号码的全部内容，希望文章能够帮你解决MapReduce实例 - partitioner 分区实现按号段统计手机号码所遇到的程序开发问题。
如果觉得靠谱客网站的内容还不错，欢迎将靠谱客网站推荐给程序员好友。
本图文内容来源于网友提供，作为学习参考使用，或来自网络收集整理，版权属于原作者所有。
发表评论 取消回复