MapReduce的手机流量统计的案例
1.程序需要的材料 文件中各个字段的含义,其中第6,7,8,9是要统计的流量相关的字段. 文件内容: 13726230503 00-FD-07-A4-72-B8:CMCC 120.196.100.82 i02.c.aliimg.com 24 27 2481 24681 200 13826544101 5C-0E-8B-C7-F1-E0:CMCC 120.197.40.4 4 0 264 0 200 13926435656 20-10-7A-28-CC-0A:CMCC 120.196.100.99 2 4 132 1512 200 13926251106 5C-0E-8B-8B-B1-50:CMCC 120.197.40.4 4 0 240 0 200 18211575961 94-71-AC-CD-E6-18:CMCC-EASY 120.196.100.99 iface.qiyi.com 视频网站 15 12 1527 2106 200 84138413 5C-0E-8B-8C-E8-20:7DaysInn 120.197.40.4 122.72.52.12 20 16 4116 1432 200 13560439658 C4-17-FE-BA-DE-D9:CMCC 120.196.100.99 18 15 1116 954 200 15920133257 5C-0E-8B-C7-BA-20:CMCC 120.197.40.4 sug.so.360.cn 信息安全 20 20 3156 2936 200 13719199419 68-A1-B7-03-07-B1:CMCC-EASY 120.196.100.82 4 0 240 0 200 13660577991 5C-0E-8B-92-5C-20:CMCC-EASY 120.197.40.4 s19.cnzz.com 站点统计 24 9 6960 690 200 15013685858 5C-0E-8B-C7-F7-90:CMCC 120.197.40.4 rank.ie.sogou.com 搜索引擎 28 27 3659 3538 200 15989002119 E8-99-C4-4E-93-E0:CMCC-EASY 120.196.100.99 www.umeng.com 站点统计 3 3 1938 180 200 13560439658 C4-17-FE-BA-DE-D9:CMCC 120.196.100.99 15 9 918 4938 200 13480253104 5C-0E-8B-C7-FC-80:CMCC-EASY 120.197.40.4 3 3 180 180 200 13602846565 5C-0E-8B-8B-B6-00:CMCC 120.197.40.4 2052.flash2-http.qq.com 综合门户 15 12 1938 2910 200 13922314466 00-FD-07-A2-EC-BA:CMCC 120.196.100.82 img.qfc.cn 12 12 3008 3720 200 13502468823 5C-0A-5B-6A-0B-D4:CMCC-EASY 120.196.100.99 y0.ifengimg.com 综合门户 57 102 7335 110349 200 18320173382 84-25-DB-4F-10-1A:CMCC-EASY 120.196.100.99 input.shouji.sogou.com 搜索引擎 21 18 9531 2412 200 13925057413 00-1F-64-E1-E6-9A:CMCC 120.196.100.55 t3.baidu.com 搜索引擎 69 63 11058 48243 200 13760778710 00-FD-07-A4-7B-08:CMCC 120.196.100.82 2 2 120 120 200 13823070001 20-7C-8F-70-68-1F:CMCC 120.196.100.99 6 3 360 180 200 13600217502 00-1F-64-E2-E8-B1:CMCC 120.196.100.55 18 138 1080 186852 200 二.程序: 1 package mapreducejob; 2 3 /** 4 * 老师给的元数据信息如下: 5 * 1363157985066 13726230503 00-FD-07-A4-72-B8:CMCC 120.196.100.82 i02.c.aliimg.com 24 27 2481 24681 200 6 * 1363157995052 13826544101 5C-0E-8B-C7-F1-E0:CMCC 120.197.40.4 4 0 264 0 200 7 * 1363157991076 13926435656 20-10-7A-28-CC-0A:CMCC 120.196.100.99 2 4 132 1512 200 8 * 第六个字段是上行数据包数. 9 * 第七个字段是下行数据包数 10 * 第八个是上行总流量 11 * 第九个是下行总流量 12 */ 13 14 import java.io.DataInput; 15 import java.io.DataOutput; 16 import java.io.IOException; 17 18 import org.apache.hadoop.conf.Configuration; 19 import org.apache.hadoop.fs.Path; 20 import org.apache.hadoop.io.LongWritable; 21 import org.apache.hadoop.io.Text; 22 import org.apache.hadoop.io.Writable; 23 import org.apache.hadoop.mapreduce.Job; 24 import org.apache.hadoop.mapreduce.Mapper; 25 import org.apache.hadoop.mapreduce.Reducer; 26 import org.apache.hadoop.mapreduce.lib.input.FileInputFormat; 27 import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat; 28 29 public class TrafficApp { 30 31 public static void main(String[] args) throws Exception { 32 Job job = Job.getInstance(new Configuration(), 33 TrafficApp.class.getSimpleName()); 34 job.setJarByClass(TrafficApp.class);// 通过jar包运行. 35 36 FileInputFormat.setInputPaths(job, args[0]);// 数据输入,指定数据源 37 38 job.setMapperClass(MyMapper.class);// 给job设置map 39 job.setMapOutputKeyClass(Text.class); 40 job.setMapOutputValueClass(TrafficWritable.class); 41 42 job.setReducerClass(MyReducer.class); 43 job.setOutputKeyClass(Text.class); 44 job.setOutputValueClass(TrafficWritable.class); 45 46 FileOutputFormat.setOutputPath(job, new Path(args[1])); 47 48 job.waitForCompletion(true);// 在集群中运行 49 } 50 51 public static class MyMapper extends 52 Mapper<LongWritable, Text, Text, TrafficWritable> { 53 // 这四个参数分别是<k1,v1>和<k2,v2> 54 // k1代表的是字节的偏移量,v1是原始数据. k2是手机号.v2是每一次的通话流量 55 Text k2 = new Text();// new 一个作为k2手机号. 56 TrafficWritable v2 = new TrafficWritable();// new 一个作为v2 57 58 @Override 59 protected void map( 60 LongWritable key, 61 Text value, 62 Mapper<LongWritable, Text, Text, TrafficWritable>.Context context) 63 throws IOException, InterruptedException { 64 String line = value.toString(); 65 String[] splited = line.split("\t");// 以制表符作为拆分符得到一个字节数组. 66 // 通过原始数据文件可以看到这个里面有11个字段,所以这个拆分的数组长度为11 67 k2.set(splited[1]);// k2是手机号,在这个数组中是第二个. 68 v2.set(splited[6], splited[7], splited[8], splited[9]);// v2是代表四个流量 69 // 对应这个被拆分数组的第6,7,8,9个. 70 context.write(k2, v2); 71 } 72 73 } 74 75 public static class MyReducer extends 76 Reducer<Text, TrafficWritable, Text, TrafficWritable> { 77 // 四个参数分别是<k2,v2> <k3,v3> 78 // k2是手机号,v2是流量TrafficWritable k3是手机号,v3是流量汇总. 79 TrafficWritable v3 = new TrafficWritable(); 80 81 @Override 82 protected void reduce( 83 Text k2, 84 Iterable<TrafficWritable> v2s, 85 Reducer<Text, TrafficWritable, Text, TrafficWritable>.Context context) 86 throws IOException, InterruptedException { 87 // reduce方法的第一个形参是k2,第二个形参是v2s,第三个形参是一个context上下文 88 // v2s是流量集合.我们在reduce方法中要做的就是把v2汇总起来变成v3. 89 long t1 = 0L; 90 long t2 = 0L; 91 long t3 = 0L; 92 long t4 = 0L; 93 for (TrafficWritable v2 : v2s) { 94 t1 += v2.t1; 95 t2 += v2.t2; 96 t3 += v2.t3; 97 t4 += v2.t4; 98 } 99 v3.set(t1, t2, t3, t4);//构造v3 100 context.write(k2, v3); 101 } 102 103 } 104 105 /** 106 * 针对流量设置一个流量类. 第六个字段是上行数据包数. 第七个字段是下行数据包数. 第八个是上行总流量. 第九个是下行总流量 107 * 108 */ 109 static class TrafficWritable implements Writable { 110 // 这个类是流量统计类,这个类包含了该手机号的上传和下载的流量 111 // 在MapReduce中的键值对中代表的是v3,有四列组成. 112 long t1; 113 long t2; 114 long t3; 115 long t4; 116 117 // 再搞一个无产的构造函数,否则容易出错 118 public TrafficWritable() { 119 } 120 121 public void set(long t1, long t2, long t3, long t4) { 122 // 赋值的方法,这个地方是传入的long类型. 123 this.t1 = t1; 124 this.t2 = t2; 125 this.t3 = t3; 126 this.t4 = t4; 127 } 128 129 public void set(String t1, String t2, String t3, String t4) { 130 // 赋值的方法,这个地方是传入的String类型. 131 this.t1 = Long.parseLong(t1); 132 this.t2 = Long.parseLong(t2); 133 this.t3 = Long.parseLong(t3); 134 this.t4 = Long.parseLong(t4); 135 } 136 137 public void readFields(DataInput in) throws IOException { 138 // 四列都通过in.readLong()读进来. 139 this.t1 = in.readLong(); 140 this.t2 = in.readLong(); 141 this.t3 = in.readLong(); 142 this.t4 = in.readLong(); 143 } 144 145 public void write(DataOutput out) throws IOException { 146 // 这个对象有四列,必须要把四列都给写出去. 147 out.writeLong(t1); 148 out.writeLong(t2); 149 out.writeLong(t3); 150 out.writeLong(t4); 151 } 152 153 public String toString() { 154 // 在Reduce阶段会用到这个方法,否则输出的是哈希编码 155 return this.t1 + "\t" + this.t2 + "\t" + this.t3 + "\t" + this.t4; 156 } 157 } 158 159 } //=============================================================== 代码二: 1 package mapreduce; 2 3 import java.io.DataInput; 4 import java.io.DataOutput; 5 import java.io.IOException; 6 7 import org.apache.hadoop.conf.Configuration; 8 import org.apache.hadoop.fs.Path; 9 import org.apache.hadoop.io.LongWritable; 10 import org.apache.hadoop.io.Text; 11 import org.apache.hadoop.io.Writable; 12 import org.apache.hadoop.mapreduce.Job; 13 import org.apache.hadoop.mapreduce.Mapper; 14 import org.apache.hadoop.mapreduce.Partitioner; 15 import org.apache.hadoop.mapreduce.Reducer; 16 import org.apache.hadoop.mapreduce.lib.input.FileInputFormat; 17 import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat; 18 19 public class TrafficApp { 20 public static void main(String[] args) throws Exception { 21 Job job = Job.getInstance(new Configuration(), TrafficApp.class.getSimpleName()); 22 job.setJarByClass(TrafficApp.class); 23 24 FileInputFormat.setInputPaths(job, args[0]); 25 26 job.setMapperClass(TrafficMapper.class); 27 job.setMapOutputKeyClass(Text.class); 28 job.setMapOutputValueClass(TrafficWritable.class); 29 30 job.setNumReduceTasks(2);//设定Reduce的数量为2 31 job.setPartitionerClass(TrafficPartitioner.class);//设定一个Partitioner的类. 32 /* 33 *Partitioner是如何实现不同的Map输出分配到不同的Reduce中? 34 *在不适用指定的Partitioner时,有 一个默认的Partitioner. 35 *就是HashPartitioner. 36 *其只有一行代码,其意思就是过来的key,不管是什么,模numberReduceTasks之后 返回值就是reduce任务的编号. 37 *numberReduceTasks的默认值是1. 任何一个数模1(取余数)都是0. 38 *这个地方0就是取编号为0的Reduce.(Reduce从0开始编号.) 39 */ 40 41 job.setReducerClass(TrafficReducer.class); 42 job.setOutputKeyClass(Text.class); 43 job.setOutputValueClass(TrafficWritable.class); 44 45 FileOutputFormat.setOutputPath(job, new Path(args[1])); 46 job.waitForCompletion(true); 47 } 48 49 public static class TrafficPartitioner extends Partitioner<Text,TrafficWritable>{//k2,v2 50 51 @Override 52 public int getPartition(Text key, TrafficWritable value,int numPartitions) { 53 long phoneNumber = Long.parseLong(key.toString()); 54 return (int)(phoneNumber%numPartitions); 55 } 56 57 } 58 59 60 /** 61 * 第一个参数是LongWritable类型是文本一行数据开头的字节数 62 * 第二个参数是文本中的一行数据 Text类型 63 * 第三个参数是要输出的手机号 Text类型 64 * 第四个参数是需要我们自定义的流量类型TrafficWritable 65 * @author ABC 66 * 67 */ 68 public static class TrafficMapper extends Mapper<LongWritable, Text, Text, TrafficWritable>{ 69 Text k2 = new Text(); 70 TrafficWritable v2 = null; 71 @Override 72 protected void map(LongWritable key,Text value, Mapper<LongWritable, Text, Text, TrafficWritable>.Context context) 73 throws IOException, InterruptedException { 74 String line = value.toString(); 75 String[] splited = line.split("\t"); 76 77 k2.set(splited[1]);//这个值对应的是手机号码 78 v2 = new TrafficWritable(splited[6], splited[7], splited[8], splited[9]); 79 context.write(k2, v2); 80 } 81 82 } 83 84 public static class TrafficReducer extends Reducer <Text, TrafficWritable, Text, TrafficWritable>{ 85 @Override 86 protected void reduce(Text k2,Iterable<TrafficWritable> v2s, 87 Reducer<Text, TrafficWritable, Text, TrafficWritable>.Context context) 88 throws IOException, InterruptedException { 89 //遍历v2s 流量都这个集合里面 90 long t1 = 0L; 91 long t2 = 0L; 92 long t3 = 0L; 93 long t4 = 0L; 94 95 for (TrafficWritable v2 : v2s) { 96 t1 += v2.getT1(); 97 t2 += v2.getT2(); 98 t3 += v2.getT3(); 99 t4 += v2.getT4(); 100 } 101 TrafficWritable v3 = new TrafficWritable(t1, t2, t3, t4); 102 context.write(k2, v3); 103 } 104 } 105 106 public static class TrafficWritable implements Writable{ 107 private long t1; 108 private long t2; 109 private long t3; 110 private long t4; 111 //写两个构造方法,一个是有参数的构造方法,一个是无参数的构造方法. 112 //必须要有 一个无参数的构造方法,否则程序运行会报错. 113 114 public TrafficWritable(){ 115 super(); 116 } 117 118 public TrafficWritable(long t1, long t2, long t3, long t4) { 119 super(); 120 this.t1 = t1; 121 this.t2 = t2; 122 this.t3 = t3; 123 this.t4 = t4; 124 } 125 //在程序中读取文本穿过来的都是字符串,所以再搞一个字符串类型的构造方法 126 public TrafficWritable(String t1, String t2, String t3, String t4) { 127 super(); 128 this.t1 = Long.parseLong(t1); 129 this.t2 = Long.parseLong(t2); 130 this.t3 = Long.parseLong(t3); 131 this.t4 = Long.parseLong(t4); 132 } 133 134 public void write(DataOutput out) throws IOException { 135 //对各个成员变量进行序列化 136 out.writeLong(t1); 137 out.writeLong(t2); 138 out.writeLong(t3); 139 out.writeLong(t4); 140 } 141 142 public void readFields(DataInput in) throws IOException { 143 //对成员变量进行反序列化 144 this.t1 = in.readLong(); 145 this.t2 = in.readLong(); 146 this.t3 = in.readLong(); 147 this.t4 = in.readLong(); 148 } 149 150 public long getT1() { 151 return t1; 152 } 153 154 public void setT1(long t1) { 155 this.t1 = t1; 156 } 157 158 public long getT2() { 159 return t2; 160 } 161 162 public void setT2(long t2) { 163 this.t2 = t2; 164 } 165 166 public long getT3() { 167 return t3; 168 } 169 170 public void setT3(long t3) { 171 this.t3 = t3; 172 } 173 174 public long getT4() { 175 return t4; 176 } 177 178 public void setT4(long t4) { 179 this.t4 = t4; 180 } 181 182 @Override 183 public String toString() { 184 return t1 + "\t" + t2 + "\t" + t3 + "\t" + t4 ; 185 } 186 187 } 188 } 本文转自SummerChill博客园博客，原文链接：http://www.cnblogs.com/DreamDrive/p/6260491.html，如需转载请自行联系原作者
