Hadoop读书笔记（六）MapReduce自定义数据类型demo

Hadoop读书笔记（一）Hadoop介绍：http://blog.csdn.net/caicongyang/article/details/39898629

Hadoop读书笔记（二）HDFS的shell操作：http://blog.csdn.net/caicongyang/article/details/41253927

Hadoop读书笔记（三）Java API操作HDFS：http://blog.csdn.net/caicongyang/article/details/41290955

Hadoop读书笔记（四）HDFS体系结构：http://blog.csdn.net/caicongyang/article/details/41322649

Hadoop读书笔记（五）MapReduce统计单词demo：http://blog.csdn.net/caicongyang/article/details/41453579

1.demo说明

从给定的日志文件中统计手机流量

2.日志文件

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
1363157985079   13823070001 20-7C-8F-70-68-1F:CMCC  120.196.100.99          6   3   360 180 200
1363157985069   13600217502 00-1F-64-E2-E8-B1:CMCC  120.196.100.55          18  138 1080    186852  200

3.代码

KpiApp.java

package mapReduce;import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
import java.net.URI;import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.Writable;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.mapreduce.lib.partition.HashPartitioner;
/*** * <p> * Title: KpiApp.java * Package mapReduce * </p>* <p>* Description: 统计流量* <p>* @author Tom.Cai* @created 2014-11-25 下午10:23:33 * @version V1.0 **/
public class KpiApp {private static final String INPUT_PATH = "hdfs://192.168.80.100:9000/wlan";private static final String OUT_PATH = "hdfs://192.168.80.100:9000/wlan_out";public static void main(String[] args) throws Exception {FileSystem fileSystem = FileSystem.get(new URI(INPUT_PATH), new Configuration());Path outPath = new Path(OUT_PATH);if (fileSystem.exists(outPath)) {fileSystem.delete(outPath, true);}Job job = new Job(new Configuration(), KpiApp.class.getSimpleName());FileInputFormat.setInputPaths(job, INPUT_PATH);job.setInputFormatClass(TextInputFormat.class);job.setMapperClass(KpiMapper.class);job.setMapOutputKeyClass(Text.class);job.setMapOutputValueClass(KpiWite.class);job.setPartitionerClass(HashPartitioner.class);job.setNumReduceTasks(1);job.setReducerClass(KpiReducer.class);job.setOutputKeyClass(Text.class);job.setOutputValueClass(KpiWite.class);FileOutputFormat.setOutputPath(job, new Path(OUT_PATH));job.setOutputFormatClass(TextOutputFormat.class);job.waitForCompletion(true);}static class KpiMapper extends Mapper<LongWritable, Text, Text, KpiWite> {@Overrideprotected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {String[] splited = value.toString().split("\t");String num = splited[1];KpiWite kpi = new KpiWite(splited[6], splited[7], splited[8], splited[9]);context.write(new Text(num), kpi);}}static class KpiReducer extends Reducer<Text, KpiWite, Text, KpiWite> {@Overrideprotected void reduce(Text key, Iterable<KpiWite> value, Context context) throws IOException, InterruptedException {long upPackNum = 0L;long downPackNum = 0L;long upPayLoad = 0L;long downPayLoad = 0L;for (KpiWite kpi : value) {upPackNum += kpi.upPackNum;downPackNum += kpi.downPackNum;upPayLoad += kpi.upPayLoad;downPayLoad += kpi.downPayLoad;}context.write(key, new KpiWite(String.valueOf(upPackNum), String.valueOf(downPackNum), String.valueOf(upPayLoad), String.valueOf(downPayLoad)));}}}class KpiWite implements Writable {long upPackNum;long downPackNum;long upPayLoad;long downPayLoad;public KpiWite() {}public KpiWite(String upPackNum, String downPackNum, String upPayLoad, String downPayLoad) {this.upPackNum = Long.parseLong(upPackNum);this.downPackNum = Long.parseLong(downPackNum);this.upPayLoad = Long.parseLong(upPayLoad);this.downPayLoad = Long.parseLong(downPayLoad);}@Overridepublic void readFields(DataInput in) throws IOException {this.upPackNum = in.readLong();this.downPackNum = in.readLong();this.upPayLoad = in.readLong();this.downPayLoad = in.readLong();}@Overridepublic void write(DataOutput out) throws IOException {out.writeLong(upPackNum);out.writeLong(downPackNum);out.writeLong(upPayLoad);out.writeLong(downPayLoad);}}

欢迎大家一起讨论学习！

有用的自己收!

记录与分享，让你我共成长！欢迎查看我的其他博客；我的博客地址：http://blog.csdn.net/caicongyang