MapReduce 应用开发

Posted 2021-02-01 wbringarden

1、MapReduce 的输入输出模型
MapReduce 中，reduce函数的输入类型必须与map 函数的输出类型一致，例如
map: (k1,v1)->list(k2,v2)
reduce:(k2,list(v2))->list(k3,v3)

MapReduce 中的常用设置：

输入数据类型由输入格式(InputFormat)设置
map 输出的key 类型通过setMapOutputKeyClass 设置

2、MapReduce 输入格式：
MapReduce 处理的输入文件一般存储在HDFS上,这些输入文件的格式多种多样，比如基于行的日志文件、二进制文件等等，而且有些文件非常之大。因此，MapReduce 定义了一些接口来读取这些文件数据。

InputFormat 接口
InputFormat 能够从一个job中得到一个split 集合，再为这个split集合配上一个合适的RecordReader来读取每个split 中的数据。

3、MapReduce 输出格式：
Hadoop 默认情况下只有一个Reduce，即输出只有一个文件。输出文件的个数有Reduce的个数一致。

OutputFormat 接口：
OutputFormat 主要用于描述数据的格式，它能够将用户提供的key/value 对写入到特定格式的文件中。

4、Combiner ：
Hadoop 框架使用Mapper 将数据处理成一个<key,value> 键值对，再在网络空间中对其进行整理(shuffile)，然后使用Reducer 处理数据并进行最终输出。

在MapReduce 框架中，Combiner 是为了避免map任务和Reduce 任务之间的无效数据传输而设置的。Hadoop 允许用户针对map任务的输出指定一个合并函数，减少传输到Reducer中的数据量。通过削减mapper 的输出数量，来减少网络带宽和Reducer 上面的负载。

可以将Combiner操作看作是在每个单独的节点上先做一次Reducer操作，其输入和输出的参数和Reducer 是一样的。

5、Partitioner:
在进行MapReduce 计算时，有时候需要将最终的输出数据分到不同的文件中，比如按照省份、性别等等来进行划分。如果想要得到多个文件，就意味着有同等数量的Reducer 任务在运行。Reducer任务的数据来自于Mapper 任务，也就是说Mapper 任务要划分数据，将不同的数据分配给给不同的Reducer 运行，Mapper 任务划分数据的过程称为Partition,负责划分数据的类为Partitioner。
MapReduce 默认的Partitioner 是HashPartitioner。一般Patitioner 先计算key 的散列值，然后对Reducer 个数进行取模运算。这种方式能随机地将整个key 空间均匀地分配给每个Reducer,同时也能确保不同Mapper 产生的key 能被分配到同一个Reducer。

6、自定义RecordReader：
RecordReader 表示以怎样的方式从分片中读取一条记录，每读取一条记录都要调用一次RecordReader 类，系统默认的RocordReader 是 LineRecorderReader,它是TextInputFormat 对应的RecordReader。SequenceFileInputFormat 对应的RecordReader 是SequenceFileRecordReader。

LineRecordReader 以每行的偏移量作为读入Map 的Key,每行读入的内容作为Map 的Value。
有时候我们需要自定义RecordReader以我们需要的方式去读取记录，比如以文件名或者行号去读取记录。

自定义RecordReader 的实现步骤：
1、继承抽象类RecordReader，实现RecordReader 的一个实例。
2、实现自定义InputFormat 类，并重写其中的CreateRecordReader()方法，返回值是自定义的RecordReader实例。
3、配置job.setInputFormatClass() 为自定义的InputFormat 实例。