Hadoop开发常用的InputFormat和OutputFormat

在用hadoop的streaming读数据时，如果输入是sequence file，如果用“-inputformat org.apache.hadoop.mapred.SequenceFileInputFormat”配置读的话，读入的数据显示的话为乱码，其实是因为读入的还是sequence file格式的，包括sequencefile的头信息在内.改为“inputformat org.apache.hadoop.mapred.SequenceFileAsTextInputFormat”即可正常读取。

以下内容摘自其他地方，对inputformat和outputformat的一个粗略的介绍：

Hadoop中的Map Reduce框架依赖InputFormat提供数据，依赖OutputFormat输出数据；每一个Map Reduce程序都离不开他们。Hadoop提供了一系列InputFormat和OutputFormat方便开发，本文介绍几种常用的。



TextInputFormat

用于读取纯文本文件，文件被分为一系列以LF或者CR结束的行，key是每一行的位置（偏移量,LongWritable类型），value是每一行的内容,Text类型。

KeyValueTextInputFormat

同样用于读取文件，如果行被分隔符（缺省是tab）分割为两部分，第一部分为key，剩下的部分为value；如果没有分隔符，整行作为 key，value为空

SequenceFileInputFormat

用于读取sequence file。 sequence file是Hadoop用于存储数据自定义格式的binary文件。它有两个子类：SequenceFileAsBinaryInputFormat，将 key和value以BytesWritable的类型读出；

SequenceFileAsTextInputFormat，将key和value以 Text的类型读出。

SequenceFileInputFilter

根据filter从sequence文件中取得部分满足条件的数据，通过setFilterClass指定Filter，内置了三种 Filter，RegexFilter取key值满足指定的正则表达式的记录；PercentFilter通过指定参数f，取记录行数%f==0的记录；MD5Filter通过指定参数f，取MD5(key)%f==0的记录。

NLineInputFormat

0.18.x新加入，可以将文件以行为单位进行split，比如文件的每一行对应一个map。得到的key是每一行的位置（偏移量,LongWritable类型），value是每一行的内容,Text类型。

CompositeInputFormat，用于多个数据源的join。

TextOutputFormat，输出到纯文本文件，格式为 key + ” ” + value。

NullOutputFormat，hadoop中的/dev/null，将输出送进黑洞。

SequenceFileOutputFormat， 输出到sequence file格式文件。

MultipleSequenceFileOutputFormat, MultipleTextOutputFormat，根据key将记录输出到不同的文件。

DBInputFormat和DBOutputFormat，从DB读取，输出到DB，预计将在0.19版本加入。

转自 http://www.cnblogs.com/xuxm2007/archive/2011/09/01/2161974.html

SequenceInputFormat和SequenceOutputFormat例子

1）输出格式为SequenceFileOutputFormat

public class SequenceFileOutputFormatDemo extends Configured implements Tool {

public static class SequenceFileOutputFormatDemoMapper extends

Mapper<LongWritable, Text, LongWritable, Text> {

public void map(LongWritable key, Text value, Context context)

throws IOException, InterruptedException {

context.write(key, value);

}

}

public static void main(String[] args) throws Exception {

int nRet = ToolRunner.run(new Configuration(),

new SequenceFileOutputFormatDemo(), args);

System.out.println(nRet);

}

@Override

public int run(String[] args) throws Exception {

// TODO Auto-generated method stub

Configuration conf = getConf();

Job job = new Job(conf, "sequence file output demo ");

job.setJarByClass(SequenceFileOutputFormatDemo.class);

FileInputFormat.addInputPaths(job, args[0]);

HdfsUtil.deleteDir(args[1]);

job.setMapperClass(SequenceFileOutputFormatDemoMapper.class);

// 因为没有reducer，所以map的输出为job的最后输出，所以需要把outputkeyclass

// outputvalueclass设置为与map的输出一致

job.setOutputKeyClass(LongWritable.class);

job.setOutputValueClass(Text.class);

// 如果不希望有reducer，设置为0

job.setNumReduceTasks(0);

// 设置输出类

job.setOutputFormatClass(SequenceFileOutputFormat.class);

// 设置sequecnfile的格式，对于sequencefile的输出格式，有多种组合方式,

//从下面的模式中选择一种，并将其余的注释掉

// 组合方式1：不压缩模式

SequenceFileOutputFormat.setOutputCompressionType(job,

CompressionType.NONE);

// 组合方式2：record压缩模式，并指定采用的压缩方式 ：默认、gzip压缩等

// SequenceFileOutputFormat.setOutputCompressionType(job,

// CompressionType.RECORD);

// SequenceFileOutputFormat.setOutputCompressorClass(job,

// DefaultCodec.class);

// 组合方式3：block压缩模式，并指定采用的压缩方式 ：默认、gzip压缩等

// SequenceFileOutputFormat.setOutputCompressionType(job,

// CompressionType.BLOCK);

// SequenceFileOutputFormat.setOutputCompressorClass(job,

// DefaultCodec.class);

SequenceFileOutputFormat.setOutputPath(job, new Path(args[1]));

int result = job.waitForCompletion(true) ? 0 : 1;

return result;

}

}

2）输入格式为SequcenFileInputFormat

public class SequenceFileInputFormatDemo extends Configured implements Tool {

public static class SequenceFileInputFormatDemoMapper extends

Mapper<LongWritable, Text, Text, NullWritable> {

public void map(LongWritable key, Text value, Context context)

throws IOException, InterruptedException {

System.out.println("key: " + key.toString() + " ; value: "

+ value.toString());

}

}

public static void main(String[] args) throws Exception {

int nRet = ToolRunner.run(new Configuration(),

new SequenceFileInputFormatDemo(), args);

System.out.println(nRet);

}

@Override

public int run(String[] args) throws Exception {

Configuration conf = getConf();

Job job = new Job(conf, "sequence file input demo");

job.setJarByClass(SequenceFileInputFormatDemo.class);

FileInputFormat.addInputPaths(job, args[0]);

HdfsUtil.deleteDir(args[1]);

FileOutputFormat.setOutputPath(job, new Path(args[1]));

job.setMapperClass(SequenceFileInputFormatDemoMapper.class);

job.setNumReduceTasks(1);

job.setOutputKeyClass(Text.class);

job.setOutputValueClass(NullWritable.class);

job.setMapOutputKeyClass(Text.class);

job.setMapOutputValueClass(Text.class);

job.setInputFormatClass(SequenceFileInputFormat.class);

int result = job.waitForCompletion(true) ? 0 : 1;

return result;

}

}