MapReduce原理及实例分析-低调大师

MapReduce原理及实例分析

2017-11-06 757

前言

由于最近开始涉及MR程序的编写，之前会一点HIVE，对MR接触不多，不论从原理还是实际操作上，都有些障碍，终于在今天柳暗花明，将这一过程记录下，与大家分享~

环境准备

在VM上搭建好LINUX虚拟机，并安装配置好HADOOP2.2.0，我这里是单节点的伪分布式

在eclipse中安装hadoop插件

对我们这种MR的新手而言，最好在本地有一个HADOOP运行环境，这样有许多好处：

如果我们每次写完MR程序，都打成JAR包上传至线上服务器上运行，那么每次MR运行的时间非常长，也许等待了许久，运行结果和我们预期不一致，又得改程序重新来一边，这会有一点痛苦！

在我们本地的HADOOP上运行MR程序非常快，就那么几秒，更加重要的是，我们可以再

本地准备输入文件去测试MR的逻辑，这对调试/开发程序非常方便！

实例及原理分析

假设，我们有这样的输入文件：

cate-a spu-1 1

cate-a spu-1 2

cate-a spu-2 3

cate-a spu-2 4

cate-a spu-3 5

cate-a spu-3 6

cate-a spu-1 7

cate-a spu-4 8

cate-a spu-4 9

cate-a spu-1 8

...

我们希望得到分cate,分spu的总和，并且取分cate分spu的TOP3

如上图示，大致描述了MAP/REDUCE的运行流程：

输入文件+InputFormat 提供给MAP

需要清楚提供给MAP的KEY1/VALUE1是什么？MAP准备输出的KEY2/VALUE2是什么？

MAP输出后，会进行分区操作，也就是决定KEY2/VALUE2发到哪些reduce上
分区由job.setPartitionerClass决定
在同一个分区内，会对KEY2进行排序，依据是job.setSortComparatorClass，

如果没有设置则根据KEY的compareTo方法
接下来进入分组阶段，会构造KEY3和VALUE迭代器
分组的依据是job.setGroupingComparatorClass，只要比较器比较的相同就在同一组
KEY3/VALUE迭代器交给reduce方法处理

步骤：

自定义KEY

KEY应该是可序列化，可比较的，只需要注意实现WritableComparable即可。

重点关注compareTo方法。

 
            @Override
           
            public 
            int 
            compareTo(Cate2SpuKey that) { 
           
            System.out.println(
            "开始对KEY进行排序..."
            ); 
           
            if
            (cate2.equals(that.getCate2())){ 
           
            return 
            spu.compareTo(that.getSpu()); 
           
            }
           
            return 
            cate2.compareTo(that.getCate2()); 
           
            }

分区

分区，是KEY的第一次比较，extends Partitioner 并提供getPartition即可。

这里根据cate分区。

分组

需要注意的是，分组类必须提供构造方法，并且重载

public int compare(WritableComparable w1, WritableComparable w2) 。这里根据cate,spu分组。

通过上述的，就可以取得分cate分spu的SUM（counts）值了。

通过eclipse hadoop插件，可以方便我们上传测试文件到HDFS，可以浏览，删除HDFS文件，更加方便的是，就像运行普通JAVA程序一样的运行/调试MR程序（不在需要打成JAR包）,让我们可以追踪MR的每一步，非常方便进行逻辑性测试~

那么怎么取分cate分spu的TOP3呢？

我们只需要把上一个MR的输出文件，作为另一个MR的输入，并且以cate+counts 为KEY ，以spu为VALUE，根据cate分区，分组，排序的话:cate相同情况下，根据counts倒序；

最后在reduce阶段取TOP3即可。

 
            @Override
           
            protected 
            void 
            reduce(Cate2CountsKey key, Iterable<Text> values, 
           
            Reducer<Cate2CountsKey, Text, Text, Text>.Context context)
           
            throws 
            IOException, InterruptedException { 
           
            System.out.println(
            "reduce..."
            ); 
           
            System.out.println(
            "VALUES迭代前... key:" 
            + key.toString()); 
           
            System.out.println(
            "VALUES迭代前... key:" 
            + key.getCounts()); 
           
            int 
            top = 
            3
            ; 
           
            for
            (Text t : values){ 
           
            if
            (top > 
            0
            ){ 
           
            System.out.println(
            "VALUES迭代中... key:" 
            + key.toString()); 
           
            System.out.println(
            "VALUES迭代中... key:" 
            + key.getCounts()); 
           
            context.write(
            new 
            Text(key.getCate2() + 
            "\t" 
            + t.toString()),  
           
            new 
            Text(key.getCounts()  
           
            + 
            ""
            )); 
           
            top--;
           
            }
           
            }
           
            System.out.println(
            "reduce over..."
            ); 
           
            }

那么到现在，分组取TOP就完成了。

一个疑问：reduce阶段中的KEY到底是什么？

在上面例子中的取TOP3的MR中，我们是以cate+counts为KEY，spu为VALUE。

cate作为分区，分组的依据，排序根据同一个cate下counts倒序。如下图所示：

那么reduce方法中的KEY是什么？

spu1,spu4,spu3...是VALUES中的，那么这个迭代器的对应KEY是什么呢？

是cate+42吗？还是其他？

在VALUES迭代过程中，这个KEY会变化吗？

我们可以看下ECLIPSE中的控制台打印输出的内容：

从打印上来看，可以分析出如下结论：

分组后，交给reduce方法处理的KEY是同一组的所有KEY的第一个KEY，并且在VALUES迭代过程中，KEY并不会重新NEW，而是利用SETTER反射的方式重新设置属性值，这样在VALUES迭代过程中取得的KEY都是与之对应的KEY了。

本文转自zfz_linux_boy 51CTO博客，原文链接：http://blog.51cto.com/zhangfengzhe/1638361，如需转载请自行联系原作者

微信关注我们

原文链接：https://yq.aliyun.com/articles/481930

转载内容版权归作者及来源网站所有！

低调大师中文资讯倾力打造互联网数据资讯、行业资源、电子商务、移动互联网、网络营销平台。持续更新报道IT业界、互联网、市场资讯、驱动更新,是最及时权威的产业资讯及硬件资讯报道平台。

HiveServer2 Impersonation

CDH4.2.0开始支持hiveserver2的impersonation,只需要设置参数hive.server2.enable.impersonation(该参数在Hive里是HIVE_SERVER2_KERBEROS_IMPERSONATION，容易被误会和kerberos一起使用)为true即可，这样可以使得提交的HQL以提交用户来运行MAPREDUCE而不是运行Hiveserver的User，那么具体是如何实现的呢：当通过ThriftCliService开启一个Session时，如果开启了上面的参数，则会调用CliService的openSessionWithImpersonation方法进而SessionManager就会生成一个HiveSession的动态代理HiveSessionImplwithUGI，并把用户的UGI传入进去： session = (HiveSession)HiveSessionProxy.getProxy(hiveSessionUgi,hiveSessionUgi.getSessionUgi()); hiveSessionUgi.setProxySe...

2017-11-06

708

前阶段发生过两次JT（CDH4.2.0）OOM的问题，导致ETL过程出错；由于是刚接手的集群参数大多是默认的，修改了JT的JVM参数相关的CMS，同时减小了retireJob的interval和cachesize看看是否奏效.发现过了三天又开始报警了，查看监控看到Old gen一直在涨根本释放不掉,估计是memory leak，做个内存dump分析吧（10GB的heap size），做了两次dump发现FileSystem$Cache部分一直在涨： Google了一下果然找到了相关的Bug：https://issues.apache.org/jira/browse/MAPREDUCE-5351；解决方案主要： 1.监控JT，符合一定阈值重启 2.修改如下参数，并写脚本手动清理job file <property> <name>keep.failed.task.files</name> <value>true</value> </property> <property> <name>keep....

2017-11-06

722

资源下载

更多资源

腾讯云软件源

为解决软件依赖安装时官方源访问速度慢的问题，腾讯云为一些软件搭建了缓存服务。您可以通过使用腾讯云软件源站来提升依赖包的安装速度。为了方便用户自由搭建服务架构，目前腾讯云软件源站支持公网访问和内网访问。

Spring

Spring框架（Spring Framework）是由Rod Johnson于2002年提出的开源Java企业级应用框架，旨在通过使用JavaBean替代传统EJB实现方式降低企业级编程开发的复杂性。该框架基于简单性、可测试性和松耦合性设计理念，提供核心容器、应用上下文、数据访问集成等模块，支持整合Hibernate、Struts等第三方框架，其适用范围不仅限于服务器端开发，绝大多数Java应用均可从中受益。

Rocky Linux

Rocky Linux（中文名：洛基）是由Gregory Kurtzer于2020年12月发起的企业级Linux发行版，作为CentOS稳定版停止维护后与RHEL（Red Hat Enterprise Linux）完全兼容的开源替代方案，由社区拥有并管理，支持x86_64、aarch64等架构。其通过重新编译RHEL源代码提供长期稳定性，采用模块化包装和SELinux安全架构，默认包含GNOME桌面环境及XFS文件系统，支持十年生命周期更新。

WebStorm

WebStorm 是jetbrains公司旗下一款JavaScript 开发工具。目前已经被广大中国JS开发者誉为“Web前端开发神器”、“最强大的HTML5编辑器”、“最智能的JavaScript IDE”等。与IntelliJ IDEA同源，继承了IntelliJ IDEA强大的JS部分的功能。