疯狂Hadoop之MapReduce工作机制(七)
MapReduce数据压缩
概述
压缩技术能够有效减少底层存储系统（HDFS）读写字节数。压缩提高了网络带宽和磁盘空间的效率。在Hadoop下，尤其是数据规模很大和工作负载密集的情况下，使用数据压缩显得非常重要。在这种情况下，I/O操作和网络数据传输要花大量的时间。还有，Shuffle与Merge过程同样也面临着巨大的I/O压力。
鉴于磁盘I/O和网络带宽是Hadoop的宝贵资源，数据压缩对于节省资源、最小化磁盘I/O和网络传输非常有帮助。不过，尽管压缩与解压操作的CPU开销不高，其性能的提升和资源的节省并非没有代价。
如果磁盘I/O和网络带宽影响了MapReduce作业性能，在任意MapReduce阶段启用压缩都可以改善端到端处理时间并减少I/O和网络流量。
压缩mapreduce的一种优化策略：通过压缩编码对mapper或者reducer的输出进行压缩，以减少磁盘IO，提高MR程序运行速度（但相应增加了cpu运算负担）
注意：压缩特性运用得当能提高性能，但运用不当也可能降低性能
基本原则：
（1）运算密集型的job，少用压缩
（2）IO密集型的job，多用压缩
MR支持的压缩编码
采用压缩的位置
压缩可以在MapReduce作用的任意阶段启用。
1）输入压缩：
在有大量数据并计划重复处理的情况下，应该考虑对输入进行压缩。然而，你无须显示指定使用的编解码方式。Hadoop自动检查文件扩展名，如果扩展名能够匹配，就会用恰当的编解码方式对文件进行压缩和解压。否则，Hadoop就不会使用任何编解码器。
2）压缩mapper输出：
当map任务输出的中间数据量很大时，应考虑在此阶段采用压缩技术。这能显著改善内部数据Shuffle过程，而Shuffle过程在Hadoop处理过程中是资源消耗最多的环节。如果发现数据量大造成网络传输缓慢，应该考虑使用压缩技术。可用于压缩mapper输出的快速编解码器包括LZO、LZ4或者Snappy。
注：LZO是供Hadoop压缩数据用的通用压缩编解码器。其设计目标是达到与硬盘读取速度相当的压缩速度，因此速度是优先考虑的因素，而不是压缩率。与gzip编解码器相比，它的压缩速度是gzip的5倍，而解压速度是gzip的2倍。同一个文件用LZO压缩后比用gzip压缩后大50%，但比压缩前小25%~50%。这对改善性能非常有利，map阶段完成时间快4倍。
3）压缩reducer输出：
在此阶段启用压缩技术能够减少要存储的数据量，因此降低所需的磁盘空间。当mapreduce作业形成作业链条时，因为第二个作业的输入也已压缩，所以启用压缩同样有效。
压缩配置参数
要在Hadoop中启用压缩，可以配置如下参数（mapred-site.xml文件中）：
计数器应用
Hadoop为每个作业维护若干内置计数器，以描述多项指标。例如，某些计数器记录已处理的字节数和记录数，使用户可监控已处理的输入数据量和已产生的输出数据量。
1）API
（1）采用枚举的方式统计计数
enum MyCounter{MALFORORMED,NORMAL}
//对枚举定义的自定义计数器加1
context.getCounter(MyCounter.MALFORORMED).increment(1);
（2）采用计数器组、计数器名称的方式统计
context.getCounter("counterGroup", "countera").increment(1);
组名和计数器名称随便起，但最好有意义。
（3）计数结果在程序运行后的控制台上查看。
MapReduce与Yarn
Yarn概述
Yarn是一个资源调度平台，负责为运算程序提供服务器运算资源，相当于一个分布式的操作系统平台，而mapreduce等运算程序则相当于运行于操作系统之上的应用程序
Yarn的重要概念
1）Yarn并不清楚用户提交的程序的运行机制
2）Yarn只提供运算资源的调度（用户程序向Yarn申请资源，Yarn就负责分配资源）
3）Yarn中的主管角色叫ResourceManager
4）Yarn中具体提供运算资源的角色叫NodeManager
5）这样一来，Yarn其实就与运行的用户程序完全解耦，就意味着Yarn上可以运行各种类型的分布式运算程序（mapreduce只是其中的一种），比如mapreduce、storm程序，spark程序……
6）所以，spark、storm等运算框架都可以整合在Yarn上运行，只要他们各自的框架中有符合Yarn规范的资源请求机制即可
7）Yarn就成为一个通用的资源调度平台，从此，企业中以前存在的各种运算集群都可以整合在一个物理集群上，提高资源利用率，方便数据共享
Yarn工作机制
1）Yarn运行机制
2）工作机制详解
（0）Mr程序提交到客户端所在的节点
（1）yarnrunner向Resourcemanager申请一个application。
（2）rm将该应用程序的资源路径返回给yarnrunner
（3）该程序将运行所需资源提交到HDFS上
（4）程序资源提交完毕后，申请运行mrAppMaster
（5）RM将用户的请求初始化成一个task
（6）其中一个NodeManager领取到task任务。
（7）该NodeManager创建容器Container，并产生MRAppmaster
（8）Container从HDFS上拷贝资源到本地
（9）MRAppmaster向RM 申请运行maptask容器
（10）RM将运行maptask任务分配给另外两个NodeManager，另两个NodeManager分别领取任务并创建容器。
（11）MR向两个接收到任务的NodeManager发送程序启动脚本，这两个NodeManager分别启动maptask，maptask对数据分区排序。
（12）MRAppmaster向RM申请2个容器，运行reduce task。
（13）reduce task向maptask获取相应分区的数据。
（14）程序运行完毕后，MR会向RM注销自己。
作业提交全过程
1）作业提交过程之YARN
2）作业提交过程之MapReduce
3）作业提交过程之读数据
4）作业提交过程之写数据
MapReduce开发总结
mapreduce在编程的时候，基本上一个固化的模式，没有太多可灵活改变的地方，除了以下几处：
1）输入数据接口：InputFormat--->FileInputFormat(文件类型数据读取的通用抽象类) DBInputFormat （数据库数据读取的通用抽象类）
默认使用的实现类是：TextInputFormat
job.setInputFormatClass(TextInputFormat.class)
TextInputFormat的功能逻辑是：一次读一行文本，然后将该行的起始偏移量作为key，行内容作为value返回
2）逻辑处理接口： Mapper
完全需要用户自己去实现其中：map() setup() clean()
3）map输出的结果在shuffle阶段会被partition以及sort，此处有两个接口可自定义：
（1）Partitioner
有默认实现 HashPartitioner，逻辑是 根据key和numReduces来返回一个分区号； key.hashCode()&Integer.MAXVALUE % numReduces
通常情况下，用默认的这个HashPartitioner就可以，如果业务上有特别的需求，可以自定义
（2）Comparable
当我们用自定义的对象作为key来输出时，就必须要实现WritableComparable接口，override其中的compareTo()方法
4）reduce端的数据分组比较接口：Groupingcomparator
reduceTask拿到输入数据（一个partition的所有数据）后，首先需要对数据进行分组，其分组的默认原则是key相同，然后对每一组kv数据调用一次reduce()方法，并且将这一组kv中的第一个kv的key作为参数传给reduce的key，将这一组数据的value的迭代器传给reduce()的values参数
利用上述这个机制，我们可以实现一个高效的分组取最大值的逻辑：
自定义一个bean对象用来封装我们的数据，然后改写其compareTo方法产生倒序排序的效果
然后自定义一个Groupingcomparator，将bean对象的分组逻辑改成按照我们的业务分组id来分组（比如订单号）
这样，我们要取的最大值就是reduce()方法中传进来key
5）逻辑处理接口：Reducer
完全需要用户自己去实现其中 reduce() setup() clean()
6）输出数据接口：OutputFormat---> 有一系列子类FileOutputformat DBoutputFormat .....
默认实现类是TextOutputFormat，功能逻辑是：将每一个KV对向目标文本文件中输出为一行
MapReduce参数优化
资源相关参数
1）以下参数是在用户自己的mr应用程序中配置就可以生效
2）应该在yarn启动之前就配置在服务器的配置文件中才能生效
3）shuffle性能优化的关键参数，应在yarn启动之前就配置好