MapReduce输入输出类型、格式及实例，mapreduce输入输出
MapReduce输入输出类型、格式及实例，mapreduce输入输出
输入格式
1、输入分片与记录
2、文件输入
3、文本输入
4、二进制输入
5、多文件输入
6、数据库格式输入
1、输入分片与记录
1、JobClient通过指定的输入文件的格式来生成数据分片InputSplit。
2、一个分片不是数据本身，而是可分片数据的引用。
3、InputFormat接口负责生成分片。
InputFormat 负责处理MR的输入部分，有三个作用:
验证作业的输入是否规范。
把输入文件切分成InputSplit。
提供RecordReader 的实现类，把InputSplit读到Mapper中进行处理。
2、文件输入
抽象类：FilelnputFormat
1、FilelnputFormat是所有使用文件作为数据源的InputFormat实现的基类。
2、FilelnputFormat输入数据格式的分片大小由数据块大小决定。
FileInputFormat保存作为job输入的所有文件，并实现了对输入文件计算splits的方法。至于获得记录的方法是有不同的子类——TextInputFormat进行实现的。
package org.apache.hadoop.mapreduce.lib.input;
public abstract class FileInputFormat<K, V> extends InputFormat<K, V> {
protected long computeSplitSize(long blockSize, long minSize,long maxSize) {
return Math.max(minSize, Math.min(maxSize, blockSize));
}
/*Generate the list of files and make them into FileSplits.*/
public List<InputSplit> getSplits(JobContext job) throws IOException {
long minSize = Math.max(getFormatMinSplitSize(), getMinSplitSize(job));
long maxSize = getMaxSplitSize(job);
......
long blockSize = file.getBlockSize();
long splitSize = computeSplitSize(blockSize, minSize, maxSize);
......
}
/*Get the minimum split size*/
public static long getMinSplitSize(JobContext job) {
return job.getConfiguration().getLong(SPLIT_MINSIZE, 1L);
}
/*Get the maximum split size.*/
public static long getMaxSplitSize(JobContext context) {
return context.getConfiguration().getLong(SPLIT_MAXSIZE,Long.MAX_VALUE);
}
//是否分片
/*
Is the given filename splitable? Usually, true, but if the file is stream compressed, it will not be.
<code>FileInputFormat</code> implementations can override this and return <code>false</code> to ensure that individual input files are never split-up so that {@link Mapper}s process entire files.
*/
protected boolean isSplitable(JobContext context, Path filename) {
return true;//默认需要分片
}
}自定义输入格式
如果我们不需要分片，那我们就需要对isSplitable方法进行重写
1、继承FileInputFormat基类。
2、重写里面的getSplits(JobContext context)方法。
3、重写createRecordReader(InputSplit split,TaskAttemptContext context)方法。
InputSplit
在执行mapreduce之前，原始数据被分割成若干split，每个split作为一个map任务的输入，在map执行过程中split会被分解成一个个记录（key-value对）， map会依次处理每一个记录。
FileInputFormat只划分比HDFS block大的文件，所以FileInputFormat划分
的结果是这个文件或者是这个文件中的一部分。
如果一个文件的大小比block小，将不会被划分，这也是Hadoop处理大文件
的效率要比处理很多小文件的效率高的原因。
当Hadoop处理很多小文件（文件大小小于hdfs block大小）的时候，由于
FileInputFormat不会对小文件进行划分，所以每一个小文件都会被当做一个split并分配一个map任务，导致效率底下。
例如：一个1G的文件，会被划分成16个64MB的split，并分配16个map任务处
理，而10000个100kb的文件会被10000个map任务处理。
Map任务的数量？
一个InputSplit对应一个Map task。
InputSplit的大小是由Math.max(minSize,Math.min(maxSize, blockSize))决定。
单节点一般10-100个map task。map task执行时长不建议低于1 分钟，否
则效率低。
抽象类：CombineFilelnputFormat
1、可以使用CombineFilelnputFormat来合并小文件。
2、因为CombineFilelnputFormat是一个抽象类，使用的时候需要创建一个
CombineFilelnputFormat的实体类，并且实现getRecordReader()的方法。
3、避免文件分法的方法：
A、数据块大小尽可能大，这样使文件的大小小于数据块的大小，就不用进行分片。（这种方式不太友好）
B、继承FilelnputFormat，并且重写isSplitable(）方法。
job.setInputFormatClass(CombineTextInputFormat.class);Hadoop2.6.0 CombineTextInputFormat源码：
package org.apache.hadoop.mapreduce.lib.input;
/* Input format that is a <code>CombineFileInputFormat</code>-equivalent for <code>TextInputFormat</code>.*/
public class CombineTextInputFormat
extends CombineFileInputFormat<LongWritable,Text> {
public RecordReader<LongWritable,Text> createRecordReader(InputSplit split,
TaskAttemptContext context) throws IOException {
return new CombineFileRecordReader<LongWritable,Text>(
(CombineFileSplit)split, context, TextRecordReaderWrapper.class);
}
/*A record reader that may be passed to <code>CombineFileRecordReader</code> so that it can be used in a <code>CombineFileInputFormat</code>-equivalent for <code>TextInputFormat</code>.*/
private static class TextRecordReaderWrapper
extends CombineFileRecordReaderWrapper<LongWritable,Text> {
// this constructor signature is required by CombineFileRecordReader
public TextRecordReaderWrapper(CombineFileSplit split,
TaskAttemptContext context, Integer idx)
throws IOException, InterruptedException {
super(new TextInputFormat(), split, context, idx);
}
}
}
3、文本输入
类名：TextlnputFormat
1、TextlnputFormat是默认的lnputFormat，每一行数据就是一条记录。
2、TextlnputFormat的key是LongWritable类型的，存储该行在整个文件的偏移量，value是每行的数据内容，Text类型。
3、输入分片与HDFS数据块关系：TextlnputFormat每一条记录就是一行，很有可能某一行跨数据块存放。默认以\n或回车键作为一行记录。
4、TextInputFormat继承了FileInputFormat。
类名：KeyValueTextInputFormat
可以通过设置key为行号的方式来知道记录的行号，并且可以通过key.value.separator.in.input设置key与value的分割符。
当输入数据的每一行是两列，并用tab分离的形式的时候，KeyValueTextInputformat处理这种格式的文件非常适合。
如果行中有分隔符，那么分隔符前面的作为key，后面的作为value；如果行中没有分隔符，那么整行作为key，value为空。
job.setInputFormatClass(KeyValueTextInputFormat.class);
//默认分隔符就是制表符
//conf.setStrings(KeyValueLineRecordReader.KEY_VALUE
_SEPERATOR, "\t")类名：NLineInputFormat
可以设置每个mapper处理的行数，可以通过mapred.line.input.format.lienspermap属性设置。
NLineInputformat可以控制在每个split中数据的行数。
//设置具体输入处理类
job.setInputFormatClass(NLineInputFormat.class);
//设置每个split的行数
NLineInputFormat.setNumLinesPerSplit(job, Integer.parseInt(args[2]));4、二进制输入
输入类：
SequenceFileInputFormat 将key和value以sequencefile格式输入。
SequenceFileAsTextInputFormat
SequenceFileAsBinaryInputFormat 将key和value以原始二进制的格式输入。由于SequenceFile能够支持Splittable，所以能够作为mapreduce输入文件的格式，能够很方便的得到己经含有＜key,value＞的分片。
SequenceFile处理、压缩处理。
5、多文件输入
类名：MultipleInputs
1、MultipleInputs能够提供多个输入数据类型。
2、通过addInputPath()方法来设置多路径。
6、数据库格式输入
类名：DBInputFormat
1、DBInputFormat是一个使用JDBC方式连接数据库，并且从关系型数据库中读取数据的一种输入格式。
2、有多个map会去连接数据库，有可能造成数据库崩溃，因此，避免过多的数据库连接。
3、HBase中的TablelnputFormat可以让MapReduce程序访问HBase表里的数据。
实例单输入路径
[root@master liguodong]# hdfs dfs -cat /input.txt
hello you
hello everybody
hello hadoop
[root@master liguodong]# hdfs dfs -text /tmp.seq
15/06/10 21:17:11 INFO bzip2.Bzip2Factory: Successfully loaded & initialized native-bzip2 library system-native
15/06/10 21:17:11 INFO compress.CodecPool: Got brand-new decompressor [.bz2]
100 apache software
99 chinese good
98 james NBA
97 index pass
96 apache software
95 chinese good
94 james NBA
93 index pass
......package mrinputformat;
import java.io.IOException;
import java.util.StringTokenizer;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.SequenceFileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
public class TestInputFormat {
public static class TokenizerMapper
extends Mapper<IntWritable, Text, Text, IntWritable>{
private final static IntWritable one = new IntWritable(1);//1
private Text word = new Text();
public void map(IntWritable key, Text value, Context context
) throws IOException, InterruptedException
{
StringTokenizer itr = new StringTokenizer(value.toString());
while (itr.hasMoreTokens()) {
word.set(itr.nextToken());
//k v
context.write(word, one);
}
}
}
public static class IntSumReducer
extends Reducer<Text,IntWritable,Text,IntWritable> {
private IntWritable result = new IntWritable();
public void reduce(Text key, Iterable<IntWritable> values,
Context context) throws IOException, InterruptedException {
int sum = 0;
for (IntWritable val : values)
{
sum += val.get();
}
result.set(sum);
context.write(key, result);
}
}
public static void main(String[] args) throws Exception {
//1、配置
Configuration conf = new Configuration();
Job job = Job.getInstance(conf, "word count");
//2、打包运行必须执行的方法
job.setJarByClass(TestInputFormat.class);
//3、输入路径
//hdfs://master:8020/tmp.seq
//hdfs://master:8020/output
FileInputFormat.addInputPath(job, new Path(args[0]));
//默认是TextInputFormat
job.setInputFormatClass(SequenceFileInputFormat.class);
//4、Map
job.setMapperClass(TokenizerMapper.class);
//5、Combiner
job.setCombinerClass(IntSumReducer.class);
//6、Reducer
job.setReducerClass(IntSumReducer.class);
job.setOutputKeyClass(Text.class);
//7、 输出路径
FileOutputFormat.setOutputPath(job, new Path(args[1]));
//8、提交作业
System.exit(job.waitForCompletion(true) ? 0 : 1);
}
}运行结果：
多输入路径方式
package mrinputformat;
import java.io.IOException;
import java.util.StringTokenizer;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.MultipleInputs;
import org.apache.hadoop.mapreduce.lib.input.SequenceFileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
public class TestInputFormat {
//采用TextInputFormat
public static class Mapper1
extends Mapper<LongWritable, Text, Text, IntWritable>{
private final static IntWritable one = new IntWritable(1);//1
private Text word = new Text();
public void map(LongWritable key, Text value, Context context
) throws IOException, InterruptedException
{
StringTokenizer itr = new StringTokenizer(value.toString());
while (itr.hasMoreTokens()) {
word.set(itr.nextToken());
//k v
context.write(word, one);
}
}
}
//SequenceFileInputFormat
public static class Mapper2
extends Mapper<IntWritable, Text, Text, IntWritable>{
private final static IntWritable one = new IntWritable(1);//1
private Text word = new Text();
public void map(IntWritable key, Text value, Context context
) throws IOException, InterruptedException
{
StringTokenizer itr = new StringTokenizer(value.toString());
while (itr.hasMoreTokens()) {
word.set(itr.nextToken());
//k v
context.write(word, one);
}
}
}
public static class IntSumReducer
extends Reducer<Text,IntWritable,Text,IntWritable> {
private IntWritable result = new IntWritable();
public void reduce(Text key, Iterable<IntWritable> values,
Context context) throws IOException, InterruptedException {
int sum = 0;
for (IntWritable val : values)
{
sum += val.get();
}
result.set(sum);
context.write(key, result);
}
}
public static void main(String[] args) throws Exception {
//1、配置
Configuration conf = new Configuration();
Job job = Job.getInstance(conf, "word count");
//2、打包运行必须执行的方法
job.setJarByClass(TestInputFormat.class);
//3、输入路径
//hdfs://master:8020/tmp.seq
//hdfs://master:8020/output
//单个输入路径
//FileInputFormat.addInputPath(job, new Path(args[0]));
//默认是TextInputFormat
//job.setInputFormatClass(SequenceFileInputFormat.class);
//4、Map
//job.setMapperClass(TokenizerMapper.class);
//多个输入路径
Path path1 = new Path("hdfs://master:8020/input.txt");
Path path2 = new Path("hdfs://master:8020/tmp.seq");
MultipleInputs.addInputPath(job, path1, TextInputFormat.class,Mapper1.class);
MultipleInputs.addInputPath(job, path2, SequenceFileInputFormat.class,Mapper2.class);
//5、Combiner
job.setCombinerClass(IntSumReducer.class);
//6、Reducer
job.setReducerClass(IntSumReducer.class);
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(IntWritable.class);
//7、 输出路径
FileOutputFormat.setOutputPath(job, new Path("hdfs://master:8020/output"));
//8、提交作业
System.exit(job.waitForCompletion(true) ? 0 : 1);
}
}运行结果：
输出格式
文本输出
TextOutputFormat
默认的输出格式，key是LongWritable，value是Text类型， key和value中间值用tab隔开的。
二进制输出
SequenceFileOutputFormat
将key和value以sequencefile格式输出。
SequenceFileAsBinaryOutputFormat
将key和value以原始二进制的格式输出。
MapFileOutputFormat
将key和value写入MapFile中。由于MapFile中的key是有序的，所以写入的时候必须保证记录是按key值顺序写入的。
多文件输出
MultipleOutputFormat
MultipleOutputs默认情况下一个reducer会产生一个输出，但是有些时候我们想一个reducer产生多个输出， MultipleOutputFormat和MultipleOutputs可以实现这个功能。
区别：MultipleOutputs可以产生不同类型的输出。
数据库格式输出
DBOutputFormat
评论暂时关闭