Hadoop MapReduce的类型和格式

wtuyss

1.MapReduce的类型
Hadoop的MapReduce函数遵循如下常规格式：
--map：(K1, V1) -> list(K2, V2)
--combine：(K2, list(V2)) -> list(K2, V2)
--partition：(K2, V2) -> integer
--reduce：(K2, list(V2)) -> list(K3, V3)
输入数据（K1，K2）的类型由输入格式进行设置。
由于JAVA的泛型存在类型擦除，其它的类型不能从mapper，reducer等直接导出，而需要手动设置。
其余设置如下表所示：

1）默认的MapReduce作业

job.setInputFormatClass(TextInputFormat.class);
job.setMapperClass(Mapper.class);
job.setMapOutputKeyClass(LongWritable.class);
job.setMapOutputValueClass(Text.class);
job.setPartitionerClass(HashPartitioner.class);
job.setNumReduceTasks(1);
job.setReducerClass(Reducer.class);
job.setOutputKeyClass(LongWritable.class);
job.setOutputValueClass(Text.class);
job.setOutputFormatClass(TextOutputFormat.class);
 HashPartitioner：对每条记录的键进行哈希操作以决定该记录属于那个分区让reduce处理。每个分区对应一个reducer任务，所以分区数等于作业的reduce的个数　　2.输入格式
　　1）输入分片和记录
　　一个输入分片(split)是由单个map处理的输入块。每一个map只处理一个输入分片。每个分片被划分为若干个记录，每条记录就是一个key/value对，map一个接一个的处理每条记录。输入分片和记录都是逻辑的，不必将他们对应到文件上。

public abstract class InputSplit {
//以字节为单位的长度
public abstract long getLength() throws IOException, InterruptedException;
//分片的存储位置
public abstract String[] getLocations() throws IOException, InterruptedException;
}
注：一个分片并不包含数据本身，而是指向数据的引用。　　长度：用来排序分片，以便优先处理最大的分片，从而最小化作业运行时间（近似贪婪算法）。
　　存储位置：供MapReduce系统使用以便将Map任务尽量放在分片数据附近。 
　　2）InputFormat
　　InputFormat负责创建输入分片并将它们分割成记录

public abstract class InputFormat<K, V> {
public abstract
List<InputSplit> getSplits(JobContext context
) throws IOException, InterruptedException;
public abstract
RecordReader<K,V> createRecordReader(InputSplit split,
TaskAttemptContext context
) throws IOException,
InterruptedException;
}
　　客户端通过调用getSpilts()方法获得分片列表。
　　JobTracker使用分片的存储位置来调度map任务。
　　Map任务把分片传给InputFormat的createRecordReader方法来获得这个分片的RecordReader。
　　RecordReader就是记录上的迭代器，map任务用一个RecordReader来生成记录的key/value对，然后再传递给map函数。

public class Mapper<KEYIN, VALUEIN, KEYOUT, VALUEOUT> {
......
public void run(Context context) throws IOException, InterruptedException {
setup(context);
try {
while (context.nextKeyValue()) {
map(context.getCurrentKey(), context.getCurrentValue(), context);
}
} finally {
cleanup(context);
}
}
}

--FileInputFormat类　　FileInputFormat是所有使用文件为数据源的InputFormat实现的基类.
　　它提供了两个功能：一个定义哪些文件包含在一个作业的输入中；一个为输入文件生成分片的实现。把分片割成记录的作业由其子类来完成。 
　　设置JobConf的输入路径：

public static void addInputPath(Job job, Path path);
public static void addInputPaths(Job job, String commaSeparatedPaths);
public static void setInputPaths(Job job, Path... inputPaths);
public static void setInputPaths(Job job, String commaSeparatedPaths);
一个路径可以表示一个文件，一个目录或是一个glob（文件模式）。　　注：目录中的内容不会被递归处理。如果包含子目录，也会被当做文件，从而产生错误。
　　设置PathFilter排除特定文件。无论设置与否，FileInputFormat会使用默认过滤器来排除隐藏文件(以.和_开头的文件)。

public static void setInputPathFilter(Job job, Class<? extends PathFilter> filter);
也可以通过配置属性来设置：
mapred.input.dir：逗号分隔的路径
mapred.input.path.filter.class：PathFilter类名
----FileInputFormat类的输入分片

分片大小的计算公式为

max(minimumSize, min(maximumSize, blockSize))
　　注：FileInputFormat定义的逻辑记录有时并不能很好地匹配HDFS的文件块，即逻辑记录的边界和HDFS的块的边界没有对齐。因此那些“本地”map可能会执行一些远程的读操作。

----CombineFileInputFormat

　　FileInputFormat生成的InputSplit是一个文件或该文件的一部分。
　　CombineFileInputFormat把多个文件打包到一个分片中以便每个mapper可以处理更多地数据。并且在决定哪些块放入同一个分片时，它会考虑节点和机架的因素，所以在典型的MapReduce作业中处理输入的速度并不会下降。
　　CombineFileInputFormat是一个抽象类，没有提供实体类，所以使用的时候要创建一个CombineFileInputFormat的具体类和实现createRecordReader()方法。
　　----避免切分
　　方法一：增加最小分片大小
　　方法二：覆盖FileInputFormat的isSplitable方法
　　----mapper中文件分片信息的相关属性

　　----TextInputFormat
　　TextInputFormat是默认的InputFormat。每条记录是一行输入。key是LongWritable类型，存储该行在整个文件中的字节偏移量，value是这行的内容，不包括任何终止符(换行符和回车符)，它是Text类型
　　----KeyValueTextInputFormat 
　　和TextInputFormat类似，KeyValueTextInputFormat也是将文件中的每行解析成一条记录，不同的是，它将每行的文本根据分隔符来拆分成键值对。默认的分隔符为制表符。可通过mapreduce.input.keyvaluelinerecordreader.key.value.separator属性设置。
　　----NLineInputFormat
　　如果希望mapper收到固定行数的输入，需要使用NLineInputFormat作为InputFormat。与TextInputFormat一样，key是文件中行的字节偏移量，值是行本身。
　　N是每个mapper收到的输入行数，默认时N=1，每个mapper会正好收到一行输入，mapreduce.input.lineinputformat.linespermap属性控制N的值。
　　----SequenceFileInputFormat
　　如果要用顺序文件数据作为MapReduce的输入，应用SequenceFileInputFormat。key和value由顺序文件决定，所以只需要保证map输入的类型匹配。
　　SequenceFileInputFormat可以读MapFile和SequenceFile。如果在处理顺序文件时遇到目录，SequenceFileInputFormat类会认为正在读MapFile，使用的是其数据文件，因此没有MapFileInputFormat类也是可以理解的。
　　----SequenceFileAsTextInputFormat和SequenceFileAsBinaryInputFormat
　　两者均是SequenceFileInputFormat的变体，前者将顺序文件(其实就是SequenceFile)的key和value转成Text对象，后者获取顺序文件的key和value作为二进制对象。
　　----多种输入MultipleInputs
　　MultipleInputs允许为每条输入路径指定InputFormat和Mapper，不同的Mapper输出的KEY/VALUE类型必须一致。
　　----数据库输入和输出
　　DBInputFormat和DBOutputFormat，HBase的TableInputFormat和TableOutputFormat。
　　运行太多的mapper数据库中读取数据可能会使数据库受不了。
　　另一种方法是使用Sqoop。
　　3.输出格式

1）TextOutputFormat
　　默认的输出格式，它本每条记录写成文本行。它的key/value可以是任意类型，因为TextOutputFormat调用toString()方法把它们转成字符串。
　　默认情况下key和value用制表符分割，可以通过mapreduce.output.textoutputformat.separator进行设置，与TextOutputFormat对应的输入格式是KeyValueTextInputFormat。
　　可以使用NullWritable来省略输出的key和value,这也会导致无分隔符输出。
　　2）SequenceFileOutputFormat & SequenceFileAsBinaryOutputFormat
　　SequenceFileOutputFormat将它的输出写为一个顺序文件。
　　SequenceFileAsBinaryOutputFormat把key/value对作为二进制格式写到一个SequenceFile容器中。
　　3）MapFileOutputFormat
　　把MapFile作为输出，MapFile中的key必需顺序添加，所以必须确保reducer输出的key已经排好序。
　　4）多个输出MultipleOutputs
　　5）延迟输出
　　FileOutputFormat的子类会产生输出文件，即使文件时空的。但是有些应用倾向于不创建空文件，此时就可以使用LazyOutputFormat了，它是一个封装输出格式，可以保证指定分区第一条记录输出时才真正的创建文件。要使用它，用JobConf和相关输出格式作为参数来调用setOutputFormatClass()方法即可。