首页 > 代码库 > Hadoop日记Day13---使用hadoop自定义类型处理手机上网日志

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

测试数据的下载地址为: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的分布式计算模型,分析一下如何实现我们的任务。首先我们有未经过处理的原始文件(相当于<k1,v1>),这个文件里存储着我需要的数据就是,那就是一个手机的流量的汇总数据(相当于<k3,v3>),而要从原始数据获得我们最终想要的数据,这中间需要经过一个过程,对原始数据进行初步加工处理,形成中间结果(相当于<k2,V2>),而<K2,V2>这时候代表什么呢?不难看出,将所有的原始数据经过map()函数的分组排序处理后,得到一个中间结果,这个中间结果是一个键值对<K2,V2>,而这里的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);//指定输出<k2,v2>的类型 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);//指定输出<k3,v3>的类型 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<LongWritable, Text, Text, KpiWritable>{ 56         protected void map(LongWritable key, Text value, org.apache.hadoop.mapreduce.Mapper<LongWritable,Text,Text,KpiWritable>.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<Text, KpiWritable, Text, KpiWritable>{ 66         /** 67          * @param    k2    表示整个文件中不同的手机号码     68          * @param    v2s    表示该手机号在不同时段的流量的集合 69          */ 70         protected void reduce(Text k2, java.lang.Iterable<KpiWritable> v2s, org.apache.hadoop.mapreduce.Reducer<Text,KpiWritable,Text,KpiWritable>.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     @Override106     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     @Override114     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     @Override122     public String toString() {123         return upPackNum + "\t" + downPackNum + "\t" + upPayLoad + "\t" + downPayLoad;124     }125 }
View Code

2 .运行结果如下

图 3.1

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