学步园

这两天在网上看了个MapReduce的多文件输出的帖子： http://blog.csdn.net/inkfish。写的不错。

我试着完成了一下。也是分为三个文件：我这三个文件，跟原作者的稍有不同。其中有些类是我原来写的，我直接拷贝过来的，所以有点不同。

My_LineRead.java

public class My_LineRead<K, V> extends RecordWriter<K, V>{<br /> private static final String utf8 = "UTF-8";<br /> private static final String colon = "----"; //划分符号<br /> private static final byte[] newline;<br /> static {<br /> try {<br /> newline = "/n".getBytes(utf8);<br /> } catch (UnsupportedEncodingException uee) {<br /> throw new IllegalArgumentException("can't find " + utf8 + " encoding");<br /> }<br /> }<br /> protected DataOutputStream out;<br /> private final byte[] keyValueSeparator;</p> <p> public My_LineRead(DataOutputStream out) {<br /> this(out, colon); //调用下面的构造函数<br /> }<br /> public My_LineRead(DataOutputStream out, String keyValueSeparator) {<br /> // TODO Auto-generated constructor stub<br /> this.out = out;<br /> try {<br /> this.keyValueSeparator = keyValueSeparator.getBytes(utf8);<br /> } catch (UnsupportedEncodingException e) {<br /> // TODO Auto-generated catch block<br /> throw new IllegalArgumentException("can't find " + utf8 + " encoding");<br /> }<br /> }<br /> @Override<br /> public void close(TaskAttemptContext arg0) throws IOException,<br /> InterruptedException {<br /> // TODO Auto-generated method stub<br /> out.close();<br /> }</p> <p> @Override<br /> public void write(K key, V value) throws IOException,<br /> InterruptedException {<br /> if (!(key == null && key instanceof NullWritable)){<br /> //如果key不为空者输出key<br /> if ((Object)key instanceof Text){<br /> Text to = (Text) key;<br /> out.write(to.getBytes(), 0, to.getLength());<br /> }<br /> else<br /> {<br /> out.write(key.toString().getBytes(utf8));<br /> }<br /> out.write(keyValueSeparator);<br /> }<br /> if (!(value == null && value instanceof NullWritable)){<br /> //如果value不为空则输出value<br /> if ((Object)value instanceof Text){<br /> Text to = (Text) value;<br /> out.write(to.getBytes(), 0, to.getLength());<br /> }<br /> else<br /> {<br /> out.write(value.toString().getBytes(utf8));<br /> }<br /> out.write(newline);<br /> }</p> <p> }<br /> }

MyMultipleOutputFormat.java //这个类，我添加了些注释便于理解

public abstract class MyMultipleOutputFormat <K extends WritableComparable<?>, V extends Writable><br /> extends FileOutputFormat<K, V> {<br /> //接口类，需要在主程序中实现generateFileNameForKeyValue来获取文件名<br /> private MultiRecordWriter writer = null;<br /> @Override<br /> public RecordWriter<K, V> getRecordWriter(TaskAttemptContext job)<br /> throws IOException, InterruptedException {<br /> // TODO Auto-generated method stub<br /> //如果第一次调用那么writer=null<br /> if (writer == null) {<br /> //getTaskOutputPath获取output路径<br /> writer = new MultiRecordWriter(job, getTaskOutputPath(job));<br /> }<br /> return writer;<br /> }<br /> private Path getTaskOutputPath(TaskAttemptContext conf) throws IOException {<br /> Path workPath = null;<br /> OutputCommitter committer = super.getOutputCommitter(conf);<br /> if (committer instanceof FileOutputCommitter) {<br /> workPath = ((FileOutputCommitter) committer).getWorkPath();<br /> } else {<br /> Path outputPath = super.getOutputPath(conf);<br /> if (outputPath == null) {<br /> throw new IOException("Undefined job output-path");<br /> }<br /> workPath = outputPath;<br /> }<br /> return workPath;<br /> }<br /> /**通过key, value, conf来确定输出文件名（含扩展名）*/<br /> //返回值就是文件名。可以根据key,value来判断<br /> protected abstract String generateFileNameForKeyValue(K key, V value, Configuration conf);</p> <p> //MultiRecordWriter类<br /> public class MultiRecordWriter extends RecordWriter<K, V> {<br /> /**RecordWriter的缓存*/<br /> private HashMap<String, RecordWriter<K, V>> recordWriters = null;<br /> private TaskAttemptContext job = null;<br /> /**输出目录*/<br /> private Path workPath = null;<br /> //构造函数<br /> public MultiRecordWriter(TaskAttemptContext job, Path workPath) {<br /> super();<br /> this.job = job;<br /> this.workPath = workPath;<br /> recordWriters = new HashMap<String, RecordWriter<K, V>>();<br /> }<br /> //关闭，应该可能是多个文件进行关闭，所有采用循环<br /> //recordWriters.values() 就是指的getBaseRecordWriter返回的值。<br /> @Override<br /> public void close(TaskAttemptContext context) throws IOException, InterruptedException {<br /> Iterator<RecordWriter<K, V>> values = this.recordWriters.values().iterator();<br /> while (values.hasNext()) {<br /> values.next().close(context);<br /> }<br /> this.recordWriters.clear();<br /> }<br /> @Override<br /> public void write(K key, V value) throws IOException, InterruptedException {<br /> //得到输出文件名<br /> String baseName = generateFileNameForKeyValue(key, value, job.getConfiguration());<br /> //如果recordWriters里没有文件名，那么就建立。否则就直接写值。<br /> RecordWriter<K, V> rw = this.recordWriters.get(baseName);<br /> if (rw == null) {<br /> rw = getBaseRecordWriter(job, baseName);<br /> //放入HashMap<br /> this.recordWriters.put(baseName, rw);<br /> }<br /> rw.write(key, value);<br /> }<br /> // ${mapred.out.dir}/_temporary/_${taskid}/${nameWithExtension}<br /> private RecordWriter<K, V> getBaseRecordWriter(TaskAttemptContext job, String baseName)<br /> throws IOException, InterruptedException {<br /> //获取配置文件<br /> Configuration conf = job.getConfiguration();<br /> //查看是否使用解码器<br /> boolean isCompressed = getCompressOutput(job);<br /> String keyValueSeparator = ",";<br /> RecordWriter<K, V> recordWriter = null;<br /> if (isCompressed) {<br /> Class<? extends CompressionCodec> codecClass = getOutputCompressorClass(job,<br /> GzipCodec.class);<br /> CompressionCodec codec = ReflectionUtils.newInstance(codecClass, conf);<br /> Path file = new Path(workPath, baseName + codec.getDefaultExtension());<br /> FSDataOutputStream fileOut = file.getFileSystem(conf).create(file, false);<br /> recordWriter = new My_LineRead<K, V>(new DataOutputStream(codec<br /> .createOutputStream(fileOut)), keyValueSeparator);<br /> }<br /> //如果不使用解码器<br /> else {<br /> Path file = new Path(workPath, baseName);<br /> FSDataOutputStream fileOut = file.getFileSystem(conf).create(file, false);<br /> //recordWriter = new My_LineRead<K, V>(fileOut, keyValueSeparator);<br /> //这里我使用的我自己的OutputFormat<br /> recordWriter = new My_LineRead<K, V>(fileOut);<br /> }<br /> return recordWriter;<br /> }<br /> }<br /> }<br />

最后就是测试类，WordCount_MulFileOut.java

public class WordCount_MulFileOut {<br /> public static class wordcountMapper extends<br /> Mapper<LongWritable, Text, Text, IntWritable>{<br /> private final static IntWritable one = new IntWritable(1);<br /> private Text word = new Text();<br /> public void map(LongWritable key, Text value, Context context)throws IOException, InterruptedException{<br /> String line = value.toString();<br /> StringTokenizer itr = new StringTokenizer(line);<br /> while(itr.hasMoreElements()){<br /> word.set(itr.nextToken());<br /> context.write(word, one);<br /> }<br /> }<br /> }<br /> public static class wordcountReduce extends<br /> Reducer<Text, IntWritable, Text, IntWritable>{<br /> public void reduce(Text key, Iterable<IntWritable>values, Context context)throws IOException, InterruptedException{<br /> int sum = 0;<br /> for (IntWritable str : values){<br /> sum += str.get();<br /> }<br /> context.write(key, new IntWritable(sum));<br /> }<br /> }<br /> public static class MyMultiple extends MyMultipleOutputFormat{</p> <p> @Override<br /> protected String generateFileNameForKeyValue(WritableComparable key,<br /> Writable value, Configuration conf) {<br /> // TODO Auto-generated method stub<br /> return "other.txt";<br /> }</p> <p> }<br /> public static void main(String args[])throws Exception{</p> <p> Configuration conf = new Configuration();</p> <p> Job job = new Job(conf, "wordcount");</p> <p> job.setJarByClass(WordCount_MulFileOut.class);</p> <p> job.setInputFormatClass(TextInputFormat.class);</p> <p> job.setOutputFormatClass(MyMultiple.class);<br /> job.setOutputKeyClass(Text.class);<br /> job.setOutputValueClass(IntWritable.class);</p> <p> job.setMapperClass(wordcountMapper.class);<br /> job.setReducerClass(wordcountReduce.class);<br /> job.setCombinerClass(wordcountReduce.class);</p> <p> FileInputFormat.setInputPaths(job, new Path(args[1]));<br /> FileOutputFormat.setOutputPath(job, new Path(args[2]));</p> <p> job.waitForCompletion(true);<br /> }<br /> }<br />