MapReduce实例 - partitioner 分区实现按号段统计手机号码

90 阅读 0 评论 60 点赞

我是靠谱客的博主凶狠白猫，这篇文章主要介绍MapReduce实例 - partitioner 分区实现按号段统计手机号码，现在分享给大家，希望可以做个参考。

一、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

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
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;
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
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）自定义分区

复制代码

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
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);
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
//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);
    }
}