一文读懂MapReduce 附流量解析实例

1.MapReduce是什么

　　 　　Hadoop MapReduce是一个软件框架，基于该框架能够容易地编写应用程序，这些应用程序能够运行在由上千个商用机器组成的大集群上，并以一种可靠的，具有容错能力的方式并行地处理上TB级别的海量数据集。这个定义里面有着这些关键词，

一是软件框架，二是并行处理，三是可靠且容错，四是大规模集群，五是海量数据集。

2 MapReduce做什么

　　 　　MapReduce擅长处理大数据，它为什么具有这种能力呢？这可由MapReduce的设计思想发觉。MapReduce的思想就是“分而治之”。

　　（1）Mapper负责“分”，即把复杂的任务分解为若干个“简单的任务”来处理。“简单的任务”包含三层含义：

一是数据或计算的规模相对原任务要大大缩小；二是就近计算原则，即任务会分配到存放着所需数据的节点上进行计算；三是这些小任务可以并行计算，彼此间几乎没有依赖关系。

　　（2）Reducer负责对map阶段的结果进行汇总。至于需要多少个Reducer，用户可以根据具体问题，通过在mapred-site.xml配置文件里设置参数mapred.reduce.tasks的值，缺省值为1。

一个比较形象的语言解释MapReduce：　　

我们要数图书馆中的所有书。你数1号书架，我数2号书架。这就是“Map”。我们人越多，数书就更快。 现在我们到一起，把所有人的统计数加在一起。这就是“Reduce”。

MapReduce流程

  • inputFormat 先通过inputFormat 读进来
  • InputSplit 然后通过split进行分片
  • RecordReaders 简称RR 通过 recordReader读取切片
  • map map处理 输出个临时结果
  • Combiner 本机先做一次reduce 减少io 提升作业执行性能，但是也有缺点，如果做全局平均数 等就不准了
  • shuffing - partitioner shuffing分发
  • shuffing - sort shuffing排序
  • reduce
  • OutputFormat 最终输出

MapReduce的输入输出

　　MapReduce框架运转在<key,value>键值对上，也就是说，框架把作业的输入看成是一组<key,value>键值对，同样也产生一组<key,value>键值对作为作业的输出，这两组键值对有可能是不同的。

　　一个MapReduce作业的输入和输出类型如下图所示：可以看出在整个流程中，会有三组<key,value>键值对类型的存在。

MapReduce的处理流程

　　这里以WordCount单词计数为例，介绍map和reduce两个阶段需要进行哪些处理。单词计数主要完成的功能是：统计一系列文本文件中每个单词出现的次数，如图所示

编写一个简单的 WordCount mapReduce 脚本

编写map脚本

//继承mapper类
/**
 *  KEYIN, Map任务读数据的key类型，offset,是每行数据起始位置的偏移量 Long
 *  VALUEIN,  Map任务读取数据的 value类型  就是一行行字符串 String
 *  KEYOUT,   map方法自定义实现输出key类型
 *  VALUEOUT  map方法自定义实现输出value类型
 *
 *  hello world welcome
 *  hello welcome
 *  keyout  String  valueout  int
 *  (world,1)
 *  hadoop 会有自定义类型  支持序列化和反序列化
 */
public class WordCountMapper extends Mapper<LongWritable,Text,Text,IntWritable> {
    //自定义map 把自己需要的数据截取出来 然后交给后续步骤来做
    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        //多个单词用 空格拆开
        String[] words = value.toString().split("-");
        for(String word:words) {
            context.write(new Text(word),new IntWritable(1));
        }
    }
}

　　

编写Reduce

/**
 * Reduce 的输入是map的输出
 *  KEYIN, VALUEIN, KEYOUT, VALUEOUT 输入是 word,1  输出是 word,3  都是 string,int
 */
public class WordCountReduce extends Reducer<Text,IntWritable,Text,IntWritable> {
    /**
     *
     * @param key  对应的单词
     * @param values  可以迭代的value 相同的key都会分发到一个reduce上面去 类似于  （hello,<1,1,1,1>）
     * @param context
     * @throws IOException
     * @throws InterruptedException
     */
    @Override
    protected void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
        int count = 0;
        Iterator<IntWritable> iterator = values.iterator();
        while(iterator.hasNext()) {
            IntWritable value = iterator.next();
            //累加
            count += value.get();
        }
        context.write(key,new IntWritable(count));
    }
}

　　

创建Job 运行

 //windows需要设置 hadoop.home.dir
    System.setProperty("hadoop.home.dir", "D:\\javaroot\\soft\\hadoop-2.6.0-cdh5.15.1");

    //设置hadoop帐号
    System.setProperty("HADOOP_USER_NAME","hadoop");
    Configuration configuration = new Configuration();
    configuration.set("fs.defaultFS","hdfs://192.168.1.100:8020");

   //提交个作业
    Job job = Job.getInstance(configuration);
    //设置job对应的主类
    job.setJarByClass(App.class);
     //添加 Combiner
    job.setCombinerClass(WordCountReduce.class);
    //设置自定义的mapper类型
    job.setMapperClass(WordCountMapper.class);
    job.setReducerClass(WordCountReduce.class);

    //设置输出类型
    job.setMapOutputKeyClass(Text.class);
    job.setMapOutputValueClass(IntWritable.class);

    //设置reduce输出
    job.setOutputKeyClass(Text.class);
    job.setOutputValueClass(IntWritable.class);

    //设置输入和输出的路径
    FileInputFormat.setInputPaths(job, new Path("/demo/wordcount/input"));
    FileOutputFormat.setOutputPath(job,new Path("/demo/wordcount/output"));

    //提交job
    boolean res  = job.waitForCompletion(true);
   System.exit(res ? 0 :1);

　　

实例：解析流量日志 算出每个手机号 上行和下行的流量和总流量

数据log

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 	13726238888	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
1363157985066 	13726238888	00-FD-07-A4-72-B8:CMCC	120.196.100.82	i02.c.aliimg.com		24	27	10000	20000	200

　　

代码实现
//map类
public class AccessMapper extends Mapper<LongWritable,Text,Text,Access> {
    //把日志按切分 找到需要的三个字段
    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        String[] lines = value.toString().split("\t");
        String phone = lines[1];
        long up = Long.parseLong(lines[lines.length - 3]);
        long down = Long.parseLong(lines[lines.length - 2]);
        context.write(new Text(phone),new Access(phone,up,down,(up+down)));
    }
}

//reduce 类
public class AccessReduce extends Reducer<Text,Access,NullWritable,Access> {
    /**
     * @param key  手机号
     * @param values   Access
     * @param context
     * @throws IOException
     * @throws InterruptedException
     */
    @Override
    protected void reduce(Text key, Iterable<Access> values, Context context) throws IOException, InterruptedException {
        long ups = 0;
        long downs = 0;
        for (Access access:values) {
            ups += access.getUp();
            downs += access.getDown();
        }
        context.write(NullWritable.get(),new Access(key.toString(),ups,downs,(ups+downs)));
    }
}

//job 执行
  public static void main(String[] args) throws Exception {
    //windows需要设置 hadoop.home.dir
    System.setProperty("hadoop.home.dir", "D:\\javaroot\\soft\\hadoop-2.6.0-cdh5.15.1");
    //设置hadoop帐号
   // System.setProperty("HADOOP_USER_NAME","hadoop");
    Configuration configuration = new Configuration();
    //configuration.set("fs.defaultFS","hdfs://192.168.1.100:8020");
    Job job = Job.getInstance(configuration);
    job.setJarByClass(App.class);
    job.setMapperClass(AccessMapper.class);
    job.setReducerClass(AccessReduce.class);
    job.setMapOutputKeyClass(Text.class);
    job.setMapOutputValueClass(Access.class);
    job.setOutputKeyClass(NullWritable.class);
    job.setOutputValueClass(Access.class);
    //设置输入和输出的路径
    FileInputFormat.setInputPaths(job, new Path("input"));
    FileOutputFormat.setOutputPath(job,new Path("output"));
    //提交job
    boolean res  = job.waitForCompletion(true);
    System.exit(res ? 0 :1);

}

　　

最后执行结果

phone=‘13480253104‘, up=180, down=180, sum=360
phone=‘13502468823‘, up=7335, down=110349, sum=117684
phone=‘13560436666‘, up=1116, down=954, sum=2070
phone=‘13560439658‘, up=2034, down=5892, sum=7926
phone=‘13602846565‘, up=1938, down=2910, sum=4848
phone=‘13660577991‘, up=6960, down=690, sum=7650
phone=‘13719199419‘, up=240, down=0, sum=240
phone=‘13726230503‘, up=2481, down=24681, sum=27162
phone=‘13726238888‘, up=12481, down=44681, sum=57162
phone=‘13760778710‘, up=120, down=120, sum=240
phone=‘13826544101‘, up=264, down=0, sum=264
phone=‘13922314466‘, up=3008, down=3720, sum=6728
phone=‘13925057413‘, up=11058, down=48243, sum=59301
phone=‘13926251106‘, up=240, down=0, sum=240
phone=‘13926435656‘, up=132, down=1512, sum=1644
phone=‘15013685858‘, up=3659, down=3538, sum=7197
phone=‘15920133257‘, up=3156, down=2936, sum=6092
phone=‘15989002119‘, up=1938, down=180, sum=2118
phone=‘18211575961‘, up=1527, down=2106, sum=3633
phone=‘18320173382‘, up=9531, down=2412, sum=11943
phone=‘84138413‘, up=4116, down=1432, sum=5548

　　

原文地址：https://www.cnblogs.com/gwyy/p/12205215.html

时间： 2024-11-10 18:02:25

一文读懂MapReduce 附流量解析实例的相关文章

[转帖]MerkleDAG全面解析 一文读懂什么是默克尔有向无环图

MerkleDAG全面解析 一文读懂什么是默克尔有向无环图 2018-08-16 15:58区块链/技术 MerkleDAG作为IPFS的核心数据结构,它融合了Merkle Tree和DAG的优点,今天阿信带大家一起来探究什么是MerkleDAG,拆分解说Merkle Tree.DAG有向无环图.MerkleDAG在IPFS中的应用. MerkleDAG树形结构图 Merkle Tree Merkle Tree是由美国计算机学家Merkle于1979年申请的专利. Merkle Tree通常也被

[转帖]一文读懂 HTTP/2

一文读懂 HTTP/2 http://support.upyun.com/hc/kb/article/1048799/ 又小拍 ? 发表于:2017年05月18日 15:34:45 ? 更新于:2017年05月24日 15:06:11 今天,HTTP 1.1 已经变成互联网中主要的协议.但是在 HTTP 协议诞生初期却被认为是简单直接的协议.1996 年在 RFC 1945 中定义了 HTTP 1.0 规范,仅 60 页,到 1999 年在 RFC 2616 定义了 HTTP 1.1,增长到了

区块链产业生态、存在问题及政策建议|一文读懂新趋势

区块链产业生态.存在问题及政策建议|一文读懂新趋势 2017-03-03 09:47:50  来源: 腾讯研究院抢沙发 摘要:从技术上来讲,区块链是一种分布式的记账方法.说到记账,我们经历了从实物记账向电子记账的演变关键词: 区块链 中国信息通信研究院与腾讯研究院区块链联合课题组 卿苏德,中国信息通信研究院区块链研究团队研究员,主要研究方向为区块链和人工智能等. 一.区块链技术原理和发展趋势 01| 区块链--一种分布式记账方法 从技术上来讲,区块链是一种分布式的记账方法.说到记账,我们经历了从

从HTTP/0.9到HTTP/2：一文读懂HTTP协议的历史演变和设计思路

本文原作者阮一峰,作者博客:ruanyifeng.com. 1.引言 HTTP 协议是最重要的互联网基础协议之一,它从最初的仅为浏览网页的目的进化到现在,已经是短连接通信的事实工业标准,最新版本 HTTP/2 更是让它再次成为技术热点. 作为即时通讯开发者来说,深刻理解HTTP协议有助于在现今复杂移动网络环境下的优化和最佳实践的开展,本文将通俗易懂的地介绍 HTTP 协议的历史演变和设计思路. 学习交流: - 即时通讯开发交流3群:185926912[推荐] - 移动端IM开发入门文章:<新手入

[转帖]一文读懂分布式架构知识体系（内含超全核心知识大图）

一文读懂分布式架构知识体系(内含超全核心知识大图) https://yq.aliyun.com/articles/721007?spm=a2c4e.11153959.0.0.2f464977X7lSdH 作者 | 晓土  阿里巴巴高级工程师 姊妹篇阅读推荐:<云原生时代,分布式系统设计必备知识图谱(内含22个知识点)> 导读:本文力求从分布式基础理论.架构设计模式.工程应用.部署运维.业界方案这几大方面,介绍基于 MSA(微服务架构)的分布式知识体系大纲,从而对 SOA 到 MSA 进化有着立

一文读懂高性能网络编程中的I/O模型

1.前言 随着互联网的发展,面对海量用户高并发业务,传统的阻塞式的服务端架构模式已经无能为力.本文(和下篇<高性能网络编程(六):一文读懂高性能网络编程中的线程模型>)旨在为大家提供有用的高性能网络编程的I/O模型概览以及网络服务进程模型的比较,以揭开设计和实现高性能网络架构的神秘面纱. 限于篇幅原因,请将本文与<高性能网络编程(六):一文读懂高性能网络编程中的线程模型>连起来读,这样会让知识更连贯. 学习交流: - 即时通讯开发交流3群:185926912[推荐] - 移动端IM

一文读懂阻塞、非阻塞、同步、异步IO

原文:一文读懂阻塞.非阻塞.同步.异步IO 介绍 在谈及网络IO的时候总避不开阻塞.非阻塞.同步.异步.IO多路复用.select.poll.epoll等这几个词语.在面试的时候也会被经常问到这几个的区别.本文就来讲一下这几个词语的含义.区别以及使用方式.Unix网络编程一书中作者给出了五种IO模型:1.BlockingIO - 阻塞IO2.NoneBlockingIO - 非阻塞IO3.IO multiplexing - IO多路复用4.signal driven IO - 信号驱动IO5.a

一文读懂前端与HTML5技术的区别！

一文读懂前端与HTML5技术的区别! 现在好多人都把HTML5和前端的概念弄混淆了.例如,H5是HTML5的简称,而现在好多人把HTML5看成了是前端开发的总称.到底前端和HTML5有哪些区别? 什么是前端开发 以一个网站为例包括网站设计.前端开发.程序开发等.网站设计就是网站的外观,平面的东西.程序开发也好理解就是功能实现.而前端开发,简单来说,就是把平面效果图转换成网页,把静态转换成动态.它的工作包括了:切图.写样式.做鼠标效果和图片切换效果等.而优秀的前端开发可以保障实现这些效果的同时,即

一文读懂UGC：互联网上的生态秘密

转载自近乎: UGC(User- Generated Content)用户原创生产内容,它是相对于PGC(Professionally-produced Content)专业生产内容的一种内容来源,简单来说,由专业个人.团队针对性输出的较为权威的内容都属于PGC,例如电视节目.报纸刊物.媒体资讯 等,而你朋友圈的图文.微博.知乎的文章.唱吧音乐.秒拍的各*,大都属于UGC. UGC产品(平台)即是以普通用户自发生产内容为基础,并由此引发的人与价值内容的关联.人与人的关联.人与商业的关联,最终为U