OutputFormat输出过程的学习

转自：http://blog.csdn.net/androidlushangderen/article/details/41278351

花了大约1周的时间，终于把MapReduce的5大阶段的源码学习结束掉了，收获不少，就算本人对Hadoop学习的一个里程碑式的纪念吧。今天花了一点点的时间，把MapReduce的最后一个阶段，输出OutputFormat给做了分析，这个过程跟InputFormat刚刚好是对着干的，二者极具对称性。为什么这么说呢，待我一一分析。

OutputFormat过程的作用就是定义数据key-value的输出格式，给你处理好后的数据，究竟以什么样的形式输出呢，才能让下次别人拿到这个文件的时候能准确的提取出里面的数据。这里，我们撇开这个话题，仅仅我知道的一些定义的数据格式的方法，比如在Redis中会有这样的设计:

[key-length][key][value-length][value][key-length][key][value-length][value]...

或者说不一定非要省空间,直接搞过分隔符

[key]   [value]\n

[key]   [value]\n

[key]   [value]\n

.....

这样逐行读取，再以空格隔开，取出里面的键值对，这么做简单是简单，就是不紧凑，空间浪费得有点多。在MapReduce的OutputFormat的有种格式用的就是这种方式。

首先必须得了解OutputFormat里面到底有什么东西:

[java] view plaincopyprint?

1. public interface OutputFormat<K, V> {
2. /**
3. * Get the {@link RecordWriter} for the given job.
4. * 获取输出记录键值记录
5. *
6. * @param ignored
7. * @param job configuration for the job whose output is being written.
8. * @param name the unique name for this part of the output.
9. * @param progress mechanism for reporting progress while writing to file.
10. * @return a {@link RecordWriter} to write the output for the job.
11. * @throws IOException
12. */
13. RecordWriter<K, V> getRecordWriter(FileSystem ignored, JobConf job,
14. String name, Progressable progress)
15. throws IOException;
16. /**
17. * Check for validity of the output-specification for the job.
18. *
19. * <p>This is to validate the output specification for the job when it is
20. * a job is submitted.  Typically checks that it does not already exist,
21. * throwing an exception when it already exists, so that output is not
22. * overwritten.</p>
23. * 作业运行之前进行的检测工作，例如配置的输出目录是否存在等
24. *
25. * @param ignored
26. * @param job job configuration.
27. * @throws IOException when output should not be attempted
28. */
29. void checkOutputSpecs(FileSystem ignored, JobConf job) throws IOException;
30. }

很简单的2个方法，RecordWriter比较重要，后面的key-value的写入操作都是根据他来完成的。但是他是一个接口，在MapReduce中，我们用的最多的他的子类是FileOutputFormat：

[java] view plaincopyprint?

1. /** A base class for {@link OutputFormat}. */
2. public abstract class FileOutputFormat<K, V> implements OutputFormat<K, V> {

他是一个抽象类，但是实现了接口中的第二个方法checkOutputSpecs()方法：

[java] view plaincopyprint?

1. public void checkOutputSpecs(FileSystem ignored, JobConf job)
2. throws FileAlreadyExistsException,
3. InvalidJobConfException, IOException {
4. // Ensure that the output directory is set and not already there
5. Path outDir = getOutputPath(job);
6. if (outDir == null && job.getNumReduceTasks() != 0) {
7. throw new InvalidJobConfException("Output directory not set in JobConf.");
8. }
9. if (outDir != null) {
10. FileSystem fs = outDir.getFileSystem(job);
11. // normalize the output directory
12. outDir = fs.makeQualified(outDir);
13. setOutputPath(job, outDir);
14. // get delegation token for the outDir‘s file system
15. TokenCache.obtainTokensForNamenodes(job.getCredentials(),
16. new Path[] {outDir}, job);
17. // check its existence
18. if (fs.exists(outDir)) {
19. //如果输出目录以及存在，则抛异常
20. throw new FileAlreadyExistsException("Output directory " + outDir +
21. " already exists");
22. }
23. }
24. }

就是检查输出目录在不在的操作。在这个类里还出现了一个辅助类：

[java] view plaincopyprint?

1. public static Path getTaskOutputPath(JobConf conf, String name)
2. throws IOException {
3. // ${mapred.out.dir}
4. Path outputPath = getOutputPath(conf);
5. if (outputPath == null) {
6. throw new IOException("Undefined job output-path");
7. }
8. //根据OutputCommitter获取输出路径
9. OutputCommitter committer = conf.getOutputCommitter();
10. Path workPath = outputPath;
11. TaskAttemptContext context = new TaskAttemptContext(conf,
12. TaskAttemptID.forName(conf.get("mapred.task.id")));
13. if (committer instanceof FileOutputCommitter) {
14. workPath = ((FileOutputCommitter)committer).getWorkPath(context,
15. outputPath);
16. }
17. // ${mapred.out.dir}/_temporary/_${taskid}/${name}
18. return new Path(workPath, name);
19. }

就是上面OutputCommiter，里面定义了很多和Task,job作业相关的方法。很多时候都会与OutputFormat合作的形式出现。他也有自己的子类实现FileOutputCommiter:

[java] view plaincopyprint?

1. public class FileOutputCommitter extends OutputCommitter {
2. public static final Log LOG = LogFactory.getLog(
3. "org.apache.hadoop.mapred.FileOutputCommitter");
4. /**
5. * Temporary directory name
6. */
7. public static final String TEMP_DIR_NAME = "_temporary";
8. public static final String SUCCEEDED_FILE_NAME = "_SUCCESS";
9. static final String SUCCESSFUL_JOB_OUTPUT_DIR_MARKER =
10. "mapreduce.fileoutputcommitter.marksuccessfuljobs";
11. public void setupJob(JobContext context) throws IOException {
12. JobConf conf = context.getJobConf();
13. Path outputPath = FileOutputFormat.getOutputPath(conf);
14. if (outputPath != null) {
15. Path tmpDir = new Path(outputPath, FileOutputCommitter.TEMP_DIR_NAME);
16. FileSystem fileSys = tmpDir.getFileSystem(conf);
17. if (!fileSys.mkdirs(tmpDir)) {
18. LOG.error("Mkdirs failed to create " + tmpDir.toString());
19. }
20. }
21. }
22. ....

在Reduce阶段的后面的写阶段，FileOutputFormat是默认的输出的类型：

[java] view plaincopyprint?

1. //获取输出的key，value
2. final RecordWriter<OUTKEY, OUTVALUE> out = new OldTrackingRecordWriter<OUTKEY, OUTVALUE>(
3. reduceOutputCounter, job, reporter, finalName);
4. OutputCollector<OUTKEY,OUTVALUE> collector =
5. new OutputCollector<OUTKEY,OUTVALUE>() {
6. public void collect(OUTKEY key, OUTVALUE value)
7. throws IOException {
8. //将处理后的key,value写入输出流中，最后写入HDFS作为最终结果
9. out.write(key, value);
10. // indicate that progress update needs to be sent
11. reporter.progress();
12. }
13. };

out就是直接发挥作用的类，但是是哪个Formtat的返回的呢，我们进入OldTrackingRecordWriter继续看：

[java] view plaincopyprint?

1. public OldTrackingRecordWriter(
2. org.apache.hadoop.mapred.Counters.Counter outputRecordCounter,
3. JobConf job, TaskReporter reporter, String finalName)
4. throws IOException {
5. this.outputRecordCounter = outputRecordCounter;
6. //默认是FileOutputFormat文件输出方式
7. this.fileOutputByteCounter = reporter
8. .getCounter(FileOutputFormat.Counter.BYTES_WRITTEN);
9. Statistics matchedStats = null;
10. if (job.getOutputFormat() instanceof FileOutputFormat) {
11. matchedStats = getFsStatistics(FileOutputFormat.getOutputPath(job), job);
12. }
13. fsStats = matchedStats;
14. FileSystem fs = FileSystem.get(job);
15. long bytesOutPrev = getOutputBytes(fsStats);
16. //从配置中获取作业的输出方式
17. this.real = job.getOutputFormat().getRecordWriter(fs, job, finalName,
18. reporter);
19. long bytesOutCurr = getOutputBytes(fsStats);
20. fileOutputByteCounter.increment(bytesOutCurr - bytesOutPrev);
21. }

果然是我们所想的那样，FileOutputFormat，但是不要忘了它的getRecordWriter()是抽象方法，调用它还必须由它的子类来实现:

[java] view plaincopyprint?

1. public abstract RecordWriter<K, V> getRecordWriter(FileSystem ignored,
2. JobConf job, String name,
3. Progressable progress)
4. throws IOException;

在这里我们举出其中在InputFormat举过的例子，TextOutputFormat,SequenceFileOutputFormat，与TextInputFormat,SequenceFileInputFormat对应。也就说说上面2个子类定义了2种截然不同的输出格式，也就返回了不一样的RecordWriter实现类.在TextOutputFormat中，他定义了一个叫LineRecordWriter的定义：

[java] view plaincopyprint?

1. public RecordWriter<K, V> getRecordWriter(FileSystem ignored,
2. JobConf job,
3. String name,
4. Progressable progress)
5. throws IOException {
6. //从配置判断输出是否要压缩
7. boolean isCompressed = getCompressOutput(job);
8. //配置中获取加在key-value的分隔符
9. String keyValueSeparator = job.get("mapred.textoutputformat.separator",
10. "\t");
11. //根据是否压缩获取相应的LineRecordWriter
12. if (!isCompressed) {
13. Path file = FileOutputFormat.getTaskOutputPath(job, name);
14. FileSystem fs = file.getFileSystem(job);
15. FSDataOutputStream fileOut = fs.create(file, progress);
16. return new LineRecordWriter<K, V>(fileOut, keyValueSeparator);
17. } else {
18. Class<? extends CompressionCodec> codecClass =
19. getOutputCompressorClass(job, GzipCodec.class);
20. // create the named codec
21. CompressionCodec codec = ReflectionUtils.newInstance(codecClass, job);
22. // build the filename including the extension
23. Path file =
24. FileOutputFormat.getTaskOutputPath(job,
25. name + codec.getDefaultExtension());
26. FileSystem fs = file.getFileSystem(job);
27. FSDataOutputStream fileOut = fs.create(file, progress);
28. return new LineRecordWriter<K, V>(new DataOutputStream
29. (codec.createOutputStream(fileOut)),
30. keyValueSeparator);
31. }

他以一个内部类的形式存在于TextOutputFormat。而在SequenceFileOutputFormat中，他的形式是怎样的呢：

[java] view plaincopyprint?

1. public RecordWriter<K, V> getRecordWriter(
2. FileSystem ignored, JobConf job,
3. String name, Progressable progress)
4. throws IOException {
5. // get the path of the temporary output file
6. Path file = FileOutputFormat.getTaskOutputPath(job, name);
7. FileSystem fs = file.getFileSystem(job);
8. CompressionCodec codec = null;
9. CompressionType compressionType = CompressionType.NONE;
10. if (getCompressOutput(job)) {
11. // find the kind of compression to do
12. compressionType = getOutputCompressionType(job);
13. // find the right codec
14. Class<? extends CompressionCodec> codecClass = getOutputCompressorClass(job,
15. DefaultCodec.class);
16. codec = ReflectionUtils.newInstance(codecClass, job);
17. }
18. final SequenceFile.Writer out =
19. SequenceFile.createWriter(fs, job, file,
20. job.getOutputKeyClass(),
21. job.getOutputValueClass(),
22. compressionType,
23. codec,
24. progress);
25. return new RecordWriter<K, V>() {
26. public void write(K key, V value)
27. throws IOException {
28. out.append(key, value);
29. }
30. public void close(Reporter reporter) throws IOException { out.close();}
31. };
32. }

关键的操作都在于SequenceFile.Writer中。有不同的RecordWriter就会有不同的写入数据的方式，这里我们举LineRecordWriter的例子。看看他的写入方法:

[java] view plaincopyprint?

1. //往输出流中写入key-value
2. public synchronized void write(K key, V value)
3. throws IOException {
4. //判断键值对是否为空
5. boolean nullKey = key == null || key instanceof NullWritable;
6. boolean nullValue = value == null || value instanceof NullWritable;
7. //如果k-v都为空，则操作失败，不写入直接返回
8. if (nullKey && nullValue) {
9. return;
10. }
11. //如果key不空，则写入key
12. if (!nullKey) {
13. writeObject(key);
14. }
15. //如果key,value都不为空，则中间写入k-v分隔符，在这里为\t空格符
16. if (!(nullKey || nullValue)) {
17. out.write(keyValueSeparator);
18. }
19. //最后写入value
20. if (!nullValue) {
21. writeObject(value);
22. }

在这个方法里，我们就能看出他的存储形式就是我刚刚在上面讲的第二种存储方式。这个方法将会在下面的代码中被执行：

[java] view plaincopyprint?

1. OutputCollector<OUTKEY,OUTVALUE> collector =
2. new OutputCollector<OUTKEY,OUTVALUE>() {
3. public void collect(OUTKEY key, OUTVALUE value)
4. throws IOException {
5. //将处理后的key,value写入输出流中，最后写入HDFS作为最终结果
6. out.write(key, value);
7. // indicate that progress update needs to be sent
8. reporter.progress();
9. }
10. };

过程可以这么理解:

collector.collect()------->out.write(key, value)------->LineRecordWriter.write(key, value)------->DataOutputStream.write(key, value).

DataOutputStream是内置于LineRecordWriter的作为里面的变量存在的。这样从Reduce末尾阶段到Output的过程也完全打通了。下面可以看看这上面涉及的完整的类目关系。

下一阶段的学习，可能或偏向于Task，Job阶段的过程分析，更加宏观过程上的一个分析。也可能会分析某个功能块的实现过程，比如Hadoop的IPC过程，据说用了很多JAVA NIO的东西。