public class TextOutputFormat<K, V> extends FileOutputFormat<K, V> {      //靜態內部類，LineRecordWriter，實現了RecordWriter。這個就是我們想要的   protected static class LineRecordWriter<K, V>     implements RecordWriter<K, V> {     private static final String utf8 = "UTF-8";     private static final byte[] newline;  //newline  是用\n標記換行     static {       try {         newline = "\n".getBytes(utf8);       } catch (UnsupportedEncodingException uee) {         throw new IllegalArgumentException("can't find " + utf8 + " encoding");       }     }     protected DataOutputStream out;     private final byte[] keyValueSeparator;     public LineRecordWriter(DataOutputStream out, String keyValueSeparator) {       this.out = out;       try {         this.keyValueSeparator = keyValueSeparator.getBytes(utf8);       } catch (UnsupportedEncodingException uee) {         throw new IllegalArgumentException("can't find " + utf8 + " encoding");       }     }     public LineRecordWriter(DataOutputStream out) {       this(out, "\t");     }     /**      * Write the object to the byte stream, handling Text as a special      * case.      * @param o the object to print      * @throws IOException if the write throws, we pass it on      */     private void writeObject(Object o) throws IOException {       if (o instanceof Text) {         Text to = (Text) o;         out.write(to.getBytes(), 0, to.getLength());       } else {         out.write(o.toString().getBytes(utf8));       }     }      //核心方法，通過這個方法把key和value寫入到hdfs檔案中     public synchronized void write(K key, V value)       throws IOException {       boolean nullKey = key == null || key instanceof NullWritable;       boolean nullValue = value == null || value instanceof NullWritable;       if (nullKey && nullValue) {//如果key和value都為null或者NullWritable就退出         return;       }       if (!nullKey) {         writeObject(key);//寫入key       }       if (!(nullKey || nullValue)) {         out.write(keyValueSeparator);//寫入分隔符，預設是\t       }       if (!nullValue) {         writeObject(value); //寫入value       }       out.write(newline); //寫入換行符     }     public synchronized void close(Reporter reporter) throws IOException {       out.close();     }   }   public RecordWriter<K, V> getRecordWriter(FileSystem ignored,                                                   JobConf job,                                                   String name,                                                   Progressable progress)     throws IOException {     boolean isCompressed = getCompressOutput(job);     String keyValueSeparator = job.get("mapreduce.output.textoutputformat.separator",                                        "\t");     if (!isCompressed) {       Path file = FileOutputFormat.getTaskOutputPath(job, name);       FileSystem fs = file.getFileSystem(job);       FSDataOutputStream fileOut = fs.create(file, progress);       return new LineRecordWriter<K, V>(fileOut, keyValueSeparator);     } else {       Class<? extends CompressionCodec> codecClass =         getOutputCompressorClass(job, GzipCodec.class);       // create the named codec       CompressionCodec codec = ReflectionUtils.newInstance(codecClass, job);       // build the filename including the extension       Path file =         FileOutputFormat.getTaskOutputPath(job,                                            name + codec.getDefaultExtension());       FileSystem fs = file.getFileSystem(job);       FSDataOutputStream fileOut = fs.create(file, progress);       return new LineRecordWriter<K, V>(new DataOutputStream                                         (codec.createOutputStream(fileOut)),                                         keyValueSeparator);     }   } }