Hadoop日记Day13---使用hadoop自定义类型处理手机上网日志

yanglgzh

测试数据的下载地址为：http://pan.baidu.com/s/1gdgSn6r


一、文件分析
　　首先可以用文本编辑器打开一个HTTP_20130313143750.dat的二进制文件，这个文件的内容是我们的手机日志，文件的内容已经经过了优化，格式比较规整，便于学习研究，感兴趣的读者可以尝试一下。
　　我从中截取文件中的一行记录内容进行分析：

　　1363157985066     13726230503    00-FD-07-A4-72-B8:CMCC    120.196.100.82    i02.c.aliimg.com      
　　2     4    27    2 481    24681    200

　　该日志文件的每个记录，一共有11个字段每个字段的含义如下图1.1所示。

　　图 1.1

二、思路分析
　　我们要统计这个文件中，同一手机号的流量汇总。而我们可以从图1.1中发现，记录中有四个字段以不同的形式表示手机的流量，这时你会想到什么呢？-----那就是面向对象的概念，我们可以自定义一个类来代表一个自定义类型去包含这几个值，用类中的属性，来表示这几个字段，来方面我们对数据的操作。
　　现在我们按照MapReduce的分布式计算模型，分析一下如何实现我们的任务。首先我们有未经过处理的原始文件(相当于)，这个文件里存储着我需要的数据就是,那就是一个手机的流量的汇总数据（相当于）,而要从原始数据获得我们最终想要的数据，这中间需要经过一个过程，对原始数据进行初步加工处理，形成中间结果(相当于),而这时候代表什么呢？不难看出，将所有的原始数据经过map()函数的分组排序处理后，得到一个中间结果，这个中间结果是一个键值对,而这里的K2应该就是电话号码，V2就是我们的自定义类型表示手机流量，最后将中间数据经过reduce()函数的归一化处理，得到我们的最终结果。

三、编程实现

1. 代码如下





  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.Reducer;
15 import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
16 import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
17 import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
18 import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
19 import org.apache.hadoop.mapreduce.lib.partition.HashPartitioner;
20
21 public class KpiApp {
22     static final String INPUT_PATH = "hdfs://hadoop:9000/wlan";
23     static final String OUT_PATH = "hdfs://hadoop:9000/out";
24     public static void main(String[] args) throws Exception{
25         final Job job = new Job(new Configuration(), KpiApp.class.getSimpleName());
26         
27         FileInputFormat.setInputPaths(job, INPUT_PATH);//1.1 指定输入文件路径
28         
29         job.setInputFormatClass(TextInputFormat.class);//指定哪个类用来格式化输入文件
30         
31         job.setMapperClass(MyMapper.class);//1.2指定自定义的Mapper类
32         
33         job.setMapOutputKeyClass(Text.class);//指定输出的类型
34         job.setMapOutputValueClass(KpiWritable.class);
35         
36         job.setPartitionerClass(HashPartitioner.class);//1.3 指定分区类
37         job.setNumReduceTasks(1);
38         
39         //1.4 TODO 排序、分区
40         
41         //1.5  TODO （可选）合并
42         
43         job.setReducerClass(MyReducer.class);//2.2 指定自定义的reduce类
44         
45         job.setOutputKeyClass(Text.class);//指定输出的类型
46         job.setOutputValueClass(KpiWritable.class);
47         
48         FileOutputFormat.setOutputPath(job, new Path(OUT_PATH));//2.3 指定输出到哪里
49         
50         job.setOutputFormatClass(TextOutputFormat.class);//设定输出文件的格式化类
51         
52         job.waitForCompletion(true);//把代码提交给JobTracker执行
53     }
54
55     static class MyMapper extends Mapper{
56         protected void map(LongWritable key, Text value, org.apache.hadoop.mapreduce.Mapper.Context context) throws IOException ,InterruptedException {
57             final String[] splited = value.toString().split("\t");
58             final String msisdn = splited[1];
59             final Text k2 = new Text(msisdn);
60             final KpiWritable v2 = new KpiWritable(splited[6],splited[7],splited[8],splited[9]);
61             context.write(k2, v2);
62         };
63     }
64     
65     static class MyReducer extends Reducer{
66         /**
67          * @param    k2    表示整个文件中不同的手机号码   
68          * @param    v2s    表示该手机号在不同时段的流量的集合
69          */
70         protected void reduce(Text k2, java.lang.Iterable v2s, org.apache.hadoop.mapreduce.Reducer.Context context) throws IOException ,InterruptedException {
71             long upPackNum = 0L;
72             long downPackNum = 0L;
73             long upPayLoad = 0L;
74             long downPayLoad = 0L;
75            
76             for (KpiWritable kpiWritable : v2s) {
77                 upPackNum += kpiWritable.upPackNum;
78                 downPackNum += kpiWritable.downPackNum;
79                 upPayLoad += kpiWritable.upPayLoad;
80                 downPayLoad += kpiWritable.downPayLoad;
81             }
82            
83             final KpiWritable v3 = new KpiWritable(upPackNum+"", downPackNum+"", upPayLoad+"", downPayLoad+"");
84             context.write(k2, v3);
85         };
86     }
87 }
88
89 class KpiWritable implements Writable{
90     long upPackNum;
91     long downPackNum;
92     long upPayLoad;
93     long downPayLoad;
94     
95     public KpiWritable(){}
96     
97     public KpiWritable(String upPackNum, String downPackNum, String upPayLoad, String downPayLoad){
98         this.upPackNum = Long.parseLong(upPackNum);
99         this.downPackNum = Long.parseLong(downPackNum);
100         this.upPayLoad = Long.parseLong(upPayLoad);
101         this.downPayLoad = Long.parseLong(downPayLoad);
102     }
103     
104     
105     @Override
106     public void readFields(DataInput in) throws IOException {
107         this.upPackNum = in.readLong();
108         this.downPackNum = in.readLong();
109         this.upPayLoad = in.readLong();
110         this.downPayLoad = in.readLong();
111     }
112
113     @Override
114     public void write(DataOutput out) throws IOException {
115         out.writeLong(upPackNum);
116         out.writeLong(downPackNum);
117         out.writeLong(upPayLoad);
118         out.writeLong(downPayLoad);
119     }
120     
121     @Override
122     public String toString() {
123         return upPackNum + "\t" + downPackNum + "\t" + upPayLoad + "\t" + downPayLoad;
124     }
125 }
View Code
2 .运行结果如下

　　图 3.1