【正文】 。 } private byte writeBuffer[] = new byte[8]。 0xFF)。 0xFF)。 0xFF)。 0xFF)。 incCount(2)。 ((v 0) amp。 } /** * Writes a codechar/code to the underlying output stream as a * 2byte value, high byte first. If no exception is thrown, the * counter codewritten/code is incremented by code2/code. * * param v a codechar/code value to be written. * exception IOException if an I/O error occurs. * see out */ public final void writeChar(int v) throws IOException { ((v 8) amp。 0xFF)。 0xFF)。 incCount(1)。 incCount(1)。 } /** * Writes a codeboolean/code to the underlying output stream as * a 1byte value. The value codetrue/code is written out as the * value code(byte)1/code。 incCount(len)。 incCount(1)。 } written = temp。 } /** * Increases the written counter by the specified value * until it reaches . */ private void incCount(int value) { int temp = written + value。 /** * bytearr is initialized on demand by writeUTF */ private byte[] bytearr = null。 }} DataOutputStream源碼DataOutputStream把基本類型轉(zhuǎn)為byte再調(diào)用out對(duì)象的write(byte b[]) write(byte b[], int off, int len)write(int b)三個(gè)方法寫入OutputStream out指向的Stream中。 } /** Returns the current position in the input. */ public int getPosition() { return ()。 } /** Resets the data that the buffer reads. */ public void reset(byte[] input, int start, int length) { (input, start, length)。 = buffer。 /** Constructs a new empty buffer. */ public DataInputBuffer() { this(new Buffer())。 } public int getLength() { return count。 } public byte[] getData() { return buf。 = start。 } public void reset(byte[] input, int start, int length) { = input。 } /** Write to a file stream */ public void writeTo(OutputStream out) throws IOException { (out)。 return this。 } /** Resets the buffer to empty. */ public DataOutputBuffer reset() { = 0。 } /** Returns the current contents of the buffer. * Data is only valid to {link getLength()}. */ public byte[] getData() { return ()。 } private DataOutputBuffer(Buffer buffer) { super(buffer)。 /** Constructs a new empty buffer. */ public DataOutputBuffer() { this(new Buffer())。 count = newcount。 buf = newbuf。 if (newcount ) { byte newbuf[] = new byte[( 1, newcount)]。 } public Buffer(int size) { super(size)。 } public int getLength() { return count。 } } 序列化框架圖 4 序列化框架體系序列化類調(diào)用過(guò)程序列化輸出OutputStream(DataOutputBuffer)SequenceFile中的Writer類：此時(shí)的OutputStream為DataOutputBuffer；MapTask收集map輸出時(shí)，序列化進(jìn)內(nèi)存，此時(shí)OutputStream為BlockingBufferIFile類的內(nèi)部類Writer類，此時(shí)的OutputStream為DataOutputBuffer；IFile類的內(nèi)部類Writer類主要用戶MapReduce過(guò)程中：map輸出、reduce的拷貝、合并階段。 } return 0i。 } Override public int pareTo(KVPair o) { int i = 。 } public void write(DataOutput out) throws IOException { ()。 } public void readFields(DataInput in) throws IOException { = ()。 } public String getWord() { return word。 public KVPair() { } public void set(String w,int c) { =c。public static class KVPair implements WritableComparableKVPair { private String word。 } } static { // register this parator (, new Comparator())。 int thatValue = readInt(b2, s2)。 } /** A Comparator optimized for IntWritable. */ public static class Comparator extends WritableComparator { public Comparator() { super()。 return (thisValuethatValue ? 1 : (thisValue==thatValue ? 0 : 1))。 } /** Compares two IntWritables. */ public int pareTo(Object o) { int thisValue = 。 return == 。 } /** Returns true iff codeo/code is a IntWritable with the same value. */ public boolean equals(Object o) { if (!(o instanceof IntWritable)) return false。 } public void readFields(DataInput in) throws IOException { value = ()。 } /** Set the value of this IntWritable. */ public void set(int value) { = value。 }Writable體系 IntWritable的實(shí)現(xiàn)代碼：public class IntWritable implements WritableComparable { private int value。 if (parator == null) parator = new WritableComparator(c, true)。 作為RawComparator實(shí)例的工廠： define方法： public static synchronized void define(Class c, WritableComparator parator) { (c, parator)。 // pare them }public int pare(WritableComparable a, WritableComparable b) { return (b)。 } catch (IOException e) { throw new RuntimeException(e)。 (b2, s2, l2)。 提供原始pare方法的默認(rèn)實(shí)現(xiàn)，能夠反序列化在流中的對(duì)象，并調(diào)用對(duì)象的pare方法：public int pare(byte[] b1, int s1, int l1, byte[] b2, int s2, int l2) { try { (b1, s1, l1)。 } // sort by key return (kvbuffer, kvindices[ii + KEYSTART], kvindices[ii + VALSTART] kvindices[ii + KEYSTART], kvbuffer, kvindices[ij + KEYSTART], kvindices[ij + VALSTART] kvindices[ij + KEYSTART])。 final int ij = kvoffsets[j % ]。RawComparator接口在MapTask的MapOutputBuffer中，其pare方法中可以看出每個(gè)參數(shù)的具體含義。圖 3壓縮繼承體系壓縮與輸入分片MapReduce中使用壓縮 序列化【優(yōu)點(diǎn)】Writable接口WritableComparable接口實(shí)現(xiàn)此接口的類可用于排序比較。 hadoop balancer Threshold 20或者 sh $HADOOP_HOME/bin/ –t 20%參數(shù)20是比例參數(shù)，表示20%，也就是平各個(gè)DataNode直接磁盤使用率偏差在20%以內(nèi)。通過(guò)構(gòu)造方法可以看到，；該類實(shí)現(xiàn)了ChecksumFileSystem類中定義但未實(shí)現(xiàn)的，用于向文件系統(tǒng)報(bào)告校驗(yàn)和文件出錯(cuò)的方法，同時(shí)把出錯(cuò)的校驗(yàn)和文件重命名后，移動(dòng)到指定的目錄（bad_files）中，在該目錄中的文件是不能夠被重新使用的文件系統(tǒng)匯總property name/name value/value/propertyproperty name/name value/value/propertyproperty name/name value/value/propertyproper