Hadoop-Mapreduce(四）

Join多种应用

• Map join（Distributedcache分布式缓存）

  • 使用场景：一张表十分小、一张表很大。

  • 解决方案

    在map端缓存多张表，提前处理业务逻辑，这样增加map端业务，减少reduce端数据的压力，尽可能的减少数据倾斜。

  • 具体办法：采用distributedcache

    ​ （1）在mapper的setup阶段，将文件读取到缓存集合中。

    ​ （2）在驱动函数中加载缓存。

    ​ job.addCacheFile(new URI(“file:/e:/mapjoincache/pd.txt”));// 缓存普通文件到task运行节点

    Reduce join

• 原理

  Map端的主要工作：为来自不同表(文件)的key/value对打标签以区别不同来源的记录。然后用连接字段作为key，其余部分和新加的标志作为value，最后进行输出。

  Reduce端的主要工作：在reduce端以连接字段作为key的分组已经完成，我们只需要在每一个分组当中将那些来源于不同文件的记录(在map阶段已经打标志)分开，最后进行合并就ok了。

• 该方法的缺点

  这种方式的缺点很明显就是会造成map和reduce端也就是shuffle阶段出现大量的数据传输，效率很低。

  数据清洗(ETL)

• 概述

在运行核心业务Mapreduce程序之前，往往要先对数据进行清洗，清理掉不符合用户要求的数据。清理的过程往往只需要运行mapper程序，不需要运行reduce程序。

计数器应用

Hadoop为每个作业维护若干内置计数器，以描述多项指标。例如，某些计数器记录已处理的字节数和记录数，使用户可监控已处理的输入数据量和已产生的输出数据量。

• API
  • 采用枚举的方式统计计数

enum MyCounter{
   MALFORORMED,NORMAL}

//对枚举定义的自定义计数器加1

context.getCounter(MyCounter.MALFORORMED).increment(1);

采用计数器组、计数器名称的方式统计

context.getCounter("counterGroup", "countera").increment(1);

• 组名和计数器名称随便起，但最好有意义。
• 计数结果在程序运行后的控制台上查看。

MapReduce开发总结

​ 在编写mapreduce程序时，需要考虑的几个方面：

• 输入数据接口：InputFormat

默认使用的实现类是：TextInputFormat 

TextInputFormat的功能逻辑是：一次读一行文本，然后将该行的起始偏移量作为key，行内容作为value返回。

KeyValueTextInputFormat每一行均为一条记录，被分隔符分割为key，value。默认分隔符是tab（\t）。

NlineInputFormat按照指定的行数N来划分切片。

CombineTextInputFormat可以把多个小文件合并成一个切片处理，提高处理效率。

用户还可以自定义InputFormat。

• 逻辑处理接口：Mapper

  用户根据业务需求实现其中三个方法：map() setup() cleanup ()

• Partitioner分区

  ​有默认实现 HashPartitioner，逻辑是根据key的哈希值和numReduces来返回一个分区号；key.hashCode()&Integer.MAXVALUE % numReduces

  ​如果业务上有特别的需求，可以自定义分区。

• Comparable排序

  ​当我们用自定义的对象作为key来输出时，就必须要实现WritableComparable接口，重写其中的compareTo()方法。

  ​部分排序：对最终输出的每一个文件进行内部排序。

  ​全排序：对所有数据进行排序，通常只有一个Reduce。

  ​二次排序：排序的条件有两个。

• Combiner合并

Combiner合并可以提高程序执行效率，减少io传输。但是使用时必须不能影响原有的业务处理结果。

• reduce端分组：Groupingcomparator

  ​reduceTask拿到输入数据（一个partition的所有数据）后，首先需要对数据进行分组，其分组的默认原则是key相同，然后对每一组kv数据调用一次reduce()方法，并且将这一组kv中的第一个kv的key作为参数传给reduce的key，将这一组数据的value的迭代器传给reduce()的values参数。

  ​利用上述这个机制，我们可以实现一个高效的分组取最大值的逻辑。

  ​自定义一个bean对象用来封装我们的数据，然后改写其compareTo方法产生倒序排序的效果。然后自定义一个Groupingcomparator，将bean对象的分组逻辑改成按照我们的业务分组id来分组（比如订单号）。这样，我们要取的最大值就是reduce()方法中传进来key。

• 逻辑处理接口：Reducer

  ​用户根据业务需求实现其中三个方法：reduce() setup() cleanup ()

• 输出数据接口：OutputFormat

  ​默认实现类是TextOutputFormat，功能逻辑是：将每一个KV对向目标文本文件中输出为一行。

SequenceFileOutputFormat将它的输出写为一个顺序文件。如果输出需要作为后续 MapReduce任务的输入，这便是一种好的输出格式，因为它的格式紧凑，很容易被压缩。

用户还可以自定义OutputFormat。