hadoop编程小技巧(7)---自定义输出文件格式以及输出到不同目录

代码测试环境:Hadoop2.4

应用场景:当需要定制输出数据格式时可以采用此技巧,包括定制输出数据的展现形式,输出路径,输出文件名称等。

Hadoop内置的输出文件格式有:

1)FileOutputFormat<K,V>  常用的父类;

2)TextOutputFormat<K,V> 默认输出字符串输出格式;

3)SequenceFileOutputFormat<K,V> 序列化文件输出;

4)MultipleOutputs<K,V> 可以把输出数据输送到不同的目录;

5) NullOutputFormat<K,V> 把输出输出到/dev/null中,即不输出任何数据,这个应用场景是在MR中进行了逻辑处理,同时输出文件已经在MR中进行了输出,而不需要在输出的情况;

6)LazyOutputFormat<K,V> 只有在调用write方法是才会产生文件,这样的话,如果没有调用write就不会产生空文件;

步骤:

类似输入数据格式,自定义输出数据格式同样可以参考下面的步骤

1) 定义一个继承自OutputFormat的类,不过一般继承FileOutputFormat即可;

2)实现其getRecordWriter方法,返回一个RecordWriter类型;

3)自定义一个继承RecordWriter的类,定义其write方法,针对每个<key,Value>写入文件数据;

实例1(修改文件默认的输出文件名以及默认的key和value的分隔符):

输入数据:

自定义CustomFileOutputFormat(把默认文件名前缀替换掉):

package fz.outputformat;

import java.io.IOException;

import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.RecordWriter;
import org.apache.hadoop.mapreduce.TaskAttemptContext;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

public class CustomOutputFormat extends FileOutputFormat<LongWritable, Text> {

	private String prefix = "custom_";
	@Override
	public RecordWriter<LongWritable, Text> getRecordWriter(TaskAttemptContext job)
			throws IOException, InterruptedException {
		// 新建一个可写入的文件
		Path outputDir = FileOutputFormat.getOutputPath(job);
//		System.out.println("outputDir.getName():"+outputDir.getName()+",otuputDir.toString():"+outputDir.toString());
		String subfix = job.getTaskAttemptID().getTaskID().toString();
		Path path = new Path(outputDir.toString()+"/"+prefix+subfix.substring(subfix.length()-5, subfix.length()));
		FSDataOutputStream fileOut = path.getFileSystem(job.getConfiguration()).create(path);
		return new CustomRecordWriter(fileOut);
	}

}

自定义CustomWriter(指定key,value分隔符):

package fz.outputformat;

import java.io.IOException;
import java.io.PrintWriter;

import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.RecordWriter;
import org.apache.hadoop.mapreduce.TaskAttemptContext;

public class CustomRecordWriter extends RecordWriter<LongWritable, Text> {

	private PrintWriter out;
	private String separator =",";
	public CustomRecordWriter(FSDataOutputStream fileOut) {
		out = new PrintWriter(fileOut);
	}

	@Override
	public void write(LongWritable key, Text value) throws IOException,
			InterruptedException {
		out.println(key.get()+separator+value.toString());
	}

	@Override
	public void close(TaskAttemptContext context) throws IOException,
			InterruptedException {
		out.close();
	}

}

调用主类:

package fz.outputformat;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
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.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

public class FileOutputFormatDriver extends Configured implements Tool{

	/**
	 * @param args
	 * @throws Exception
	 */
	public static void main(String[] args) throws Exception {
		// TODO Auto-generated method stub
		ToolRunner.run(new Configuration(), new FileOutputFormatDriver(),args);
	}

	@Override
	public int run(String[] arg0) throws Exception {
		if(arg0.length!=3){
			System.err.println("Usage:\nfz.outputformat.FileOutputFormatDriver <in> <out> <numReducer>");
			return -1;
		}
		Configuration conf = getConf();

		Path in = new Path(arg0[0]);
		Path out= new Path(arg0[1]);
		boolean delete=out.getFileSystem(conf).delete(out, true);
		System.out.println("deleted "+out+"?"+delete);
		Job job = Job.getInstance(conf,"fileouttputformat test job");
		job.setJarByClass(getClass());

		job.setInputFormatClass(TextInputFormat.class);
		job.setOutputFormatClass(CustomOutputFormat.class);

		job.setMapperClass(Mapper.class);
		job.setMapOutputKeyClass(LongWritable.class);
		job.setMapOutputValueClass(Text.class);
		job.setOutputKeyClass(LongWritable.class);
		job.setOutputValueClass(Text.class);
		job.setNumReduceTasks(Integer.parseInt(arg0[2]));
		job.setReducerClass(Reducer.class);
		FileInputFormat.setInputPaths(job, in);
		FileOutputFormat.setOutputPath(job, out);

		return job.waitForCompletion(true)?0:-1;
	}

}

查看输出:

从输出结果可以看到输出格式以及文件名确实按照预想输出了。

实例2(根据key和value值输出数据到不同目录):
自定义主类(主类其实就是修改了输出的方式而已):

package fz.multipleoutputformat;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
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.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.MultipleOutputs;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

public class FileOutputFormatDriver extends Configured implements Tool{

	/**
	 * @param args
	 * @throws Exception
	 */
	public static void main(String[] args) throws Exception {
		// TODO Auto-generated method stub
		ToolRunner.run(new Configuration(), new FileOutputFormatDriver(),args);
	}

	@Override
	public int run(String[] arg0) throws Exception {
		if(arg0.length!=3){
			System.err.println("Usage:\nfz.multipleoutputformat.FileOutputFormatDriver <in> <out> <numReducer>");
			return -1;
		}
		Configuration conf = getConf();

		Path in = new Path(arg0[0]);
		Path out= new Path(arg0[1]);
		boolean delete=out.getFileSystem(conf).delete(out, true);
		System.out.println("deleted "+out+"?"+delete);
		Job job = Job.getInstance(conf,"fileouttputformat test job");
		job.setJarByClass(getClass());

		job.setInputFormatClass(TextInputFormat.class);
//		job.setOutputFormatClass(CustomOutputFormat.class);
		MultipleOutputs.addNamedOutput(job, "ignore", TextOutputFormat.class,
				LongWritable.class, Text.class);
		MultipleOutputs.addNamedOutput(job, "other", TextOutputFormat.class,
				LongWritable.class, Text.class);

		job.setMapperClass(Mapper.class);
		job.setMapOutputKeyClass(LongWritable.class);
		job.setMapOutputValueClass(Text.class);
		job.setOutputKeyClass(LongWritable.class);
		job.setOutputValueClass(Text.class);
		job.setNumReduceTasks(Integer.parseInt(arg0[2]));
		job.setReducerClass(MultipleReducer.class);
		FileInputFormat.setInputPaths(job, in);
		FileOutputFormat.setOutputPath(job, out);

		return job.waitForCompletion(true)?0:-1;
	}

}

自定义reducer(因为要根据key和value的值输出数据到不同目录,所以需要自定义逻辑)

package fz.multipleoutputformat;

import java.io.IOException;

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.output.MultipleOutputs;

public class MultipleReducer extends
		Reducer<LongWritable, Text, LongWritable, Text> {
	private MultipleOutputs<LongWritable,Text> out;
	@Override
	public void setup(Context cxt){
		out = new MultipleOutputs<LongWritable,Text>(cxt);
	}
	@Override
	public void reduce(LongWritable key ,Iterable<Text> value,Context cxt)throws IOException,InterruptedException{
		for(Text v:value){
			if(v.toString().startsWith("ignore")){
//				System.out.println("ignore--------------------value:"+v);
				out.write("ignore", key, v, "ign");
			}else{
//				System.out.println("other---------------------value:"+v);
				out.write("other", key, v, "oth");
			}
		}
	}

	@Override
	public void cleanup(Context cxt)throws IOException,InterruptedException{
		out.close();
	}
}

查看输出:

可以看到输出的数据确实根据value的不同值被写入了不同的文件目录中,但是这里同样可以看到有默认的文件生成,同时注意到这个文件的大小是0,这个暂时还没解决。

总结:自定义输出格式,可以定制一些特殊需求,不过一般使用Hadoop内置的输出格式即可,这点来说其应用意义不是很大。不过使用Hadoop内置的MultipleOutputs可以根据数据的不同特性输出到不同的目录,还是很有实际意义的。

分享,成长,快乐

转载请注明blog地址:http://blog.csdn.net/fansy1990

hadoop编程小技巧(7)---自定义输出文件格式以及输出到不同目录

时间: 2024-10-10 09:35:44

hadoop编程小技巧(7)---自定义输出文件格式以及输出到不同目录的相关文章

hadoop编程小技巧(7)---自己定义输出文件格式以及输出到不同文件夹

代码測试环境:Hadoop2.4 应用场景:当须要定制输出数据格式时能够採用此技巧,包含定制输出数据的展现形式.输出路径.输出文件名称称等. Hadoop内置的输出文件格式有: 1)FileOutputFormat<K,V>  经常使用的父类. 2)TextOutputFormat<K,V> 默认输出字符串输出格式. 3)SequenceFileOutputFormat<K,V> 序列化文件输出: 4)MultipleOutputs<K,V> 能够把输出数据

hadoop编程小技巧(5)---自定义输入文件格式类InputFormat

Hadoop代码测试环境:Hadoop2.4 应用:在对数据需要进行一定条件的过滤和简单处理的时候可以使用自定义输入文件格式类. Hadoop内置的输入文件格式类有: 1)FileInputFormat<K,V>这个是基本的父类,我们自定义就直接使用它作为父类: 2)TextInputFormat<LongWritable,Text>这个是默认的数据格式类,我们一般编程,如果没有特别指定的话,一般都使用的是这个:key代表当前行数据距离文件开始的距离,value代码当前行字符串:

hadoop编程小技巧(3)---自定义分区类Partitioner

Hadoop代码测试环境:Hadoop2.4 原理:在Hadoop的MapReduce过程中,Mapper读取处理完成数据后,会把数据发送到Partitioner,由Partitioner来决定每条记录应该送往哪个reducer节点,默认使用的是HashPartitioner,其核心代码如下: /** Use {@link Object#hashCode()} to partition. */ public int getPartition(K2 key, V2 value, int numRe

hadoop编程小技巧(4)---全局key排序类TotalOrderPartitioner

Hadoop代码测试版本:Hadoop2.4 原理:在进行MR程序之前对输入数据进行随机提取样本,把样本排序,然后在MR的中间过程Partition的时候使用这个样本排序的值进行分组数据,这样就可以达到全局排序的目的了. 难点:如果使用Hadoop提供的方法来实现全局排序,那么要求Mapper的输入.输出的key不变才可以,因为在源码InputSampler中提供的随机抽取的数据是输入数据最原始的key,如下代码(line:225): for (int i = 0; i < splitsToSa

hadoop编程小技巧(2)---计数器Counter

Hadoop代码测试版本:2.4 应用场景:在Hadoop编程的时候,有时我们在进行我们算法逻辑的时候想附带了解下数据的一些特性,比如全部数据的记录数有多少,map的输出有多少等等信息(这些是在算法运行完毕后,直接有的),就可以使用计数器Counter. 如果是针对很特定的数据的一些统计,比如统计以1开头的所有记录数等等信息,这时就需要自定义Counter.自定义Counter有两种方式,第一种,定义枚举类型,类似: public enum MyCounters{ ALL_RECORDS,ONE

hadoop编程小技巧(9)---二次排序(值排序)

代码测试环境:Hadoop2.4 应用场景:在Reducer端一般是key排序,而没有value排序,如果想对value进行排序,则可以使用此技巧. 应用实例描述: 比如针对下面的数据: a,5 b,7 c,2 c,9 a,3 a,1 b,10 b,3 c,1 如果使用一般的MR的话,其输出可能是这样的: a 1 a 3 a 5 b 3 b 10 b 7 c 1 c 9 c 2 从数据中可以看到其键是排序的,但是其值不是.通过此篇介绍的技巧可以做到下面的输出: a 1 a 3 a 5 b 3 b

hadoop编程小技巧(6)---处理大量小数据文件CombineFileInputFormat应用

代码测试环境:Hadoop2.4 应用场景:当需要处理很多小数据文件的时候,可以应用此技巧来达到高效处理数据的目的. 原理:应用CombineFileInputFormat,可以把多个小数据文件在进行分片的时候合并.由于每个分片会产生一个Mapper,当一个Mapper处理的数据比较小的时候,其效率较低.而一般使用Hadoop处理数据时,即默认方式,会把一个输入数据文件当做一个分片,这样当输入文件较小时就会出现效率低下的情况. 实例: 参考前篇blog:hadoop编程小技巧(5)---自定义输

hadoop编程小技巧(1)---map端聚合

测试hadoop版本:2.4 Map端聚合的应用场景:当我们只关心所有数据中的部分数据时,并且数据可以放入内存中. 使用的好处:可以大大减小网络数据的传输量,提高效率: 一般编程思路:在Mapper的map函数中读入所有数据,然后添加到一个List(队列)中,然后在cleanup函数中对list进行处理,输出我们关系的少量数据. 实例: 在map函数中使用空格分隔每行数据,然后把每个单词添加到一个堆栈中,在cleanup函数中输出堆栈中单词次数比较多的单词以及次数: package fz.inm

hadoop编程小技巧(8)---Unit Testing (单元测试)

所需环境: Hadoop相关jar包(下载官网发行版即可): 下载junit包(最新为好): 下载mockito包: 下载mrunit包: 下载powermock-mockito包: 相关包截图如下(相关下载参考:http://download.csdn.net/detail/fansy1990/7690977): 应用场景: 在进行Hadoop的一般MR编程时,需要验证我们的业务逻辑,或者说是验证数据流的时候可以使用此环境,这个环境不要求真实的云平台,只是针对算法或者代码逻辑进行验证,方便调试