最近一直在找工作，寫論文，對MapReduce原始碼的學習擱置了很久，想來想去認為不能放棄，有意義的事情一定要做好，要做到底，要盡力。前面的文章到後來寫的有些心不在焉，有應付之嫌，如今重新拾起，認真學習，認真寫下去。MR 2.0已經發布很久了，新架構新思想很值得學習，學無止境啊。

參考書目：

【1】《Java程式設計思想（第四版）》

【2】《Hadoop 技術內幕：深入解析MapReduce架構設計與實現原理》

【3】《Hadoop 技術內幕：深入解析Hadoop Common和HDFS架構設計與實現原理》

從Map階段的Collect過程開始吧，力求有所收穫。

看MapTask類的入口函式run，根據配置判斷啟用old mapper或new mapper。如果是前者：

準備一個MapRunner跑使用者的map函式。這個MapRunner實現了MapRunnable泛型介面，四個泛型引數分別代表map的輸入鍵值對和輸出鍵值對的型別（INKEY,INVALUE,OUTKEY,OUTVALUE）。對於MapRunner來說，兩個泛型引數來自RecordReader<INKEY,INVALUE>物件；另外兩個來自OldOutputCollector物件。後者使用MapOutputBuffer<OUTKEY,OUTVALUE>物件構造。MapOutputBuffer類實現了泛型介面MapOutputCollector，因而具有collect功能。這樣一個MapRunner就具備了讀取資料（read）和輸出資料（collect）的功能。MapRunner通過run函式使用上述兩個功能物件。

如果是後者，就有了新的一套RecordReader和OutputCollector，並使用了一個Context物件封裝上述功能，傳入run函式。不打算詳細學習這部分內容。

回到old mapper的實現中，前面提到泛型，由於對java中的泛型技術比較陌生，這裡詳細學習一下MapRunner.run方法中涉及到泛型技術，順便還有反射的內容。

首先，看RecordReader，它是一個泛型介面。使用泛型而不是普通介面的好處是，實現介面不僅僅具有了介面的功能，同時介面方法的引數和返回值支援多種型別。對於RecordReader來說，支援多種型別的key和value。如果不使用泛型，則在介面中使用key和value型別的基類，這樣就只支援基類及其派生類，不支援該派生體系外的型別。Java支援泛型方法，這使得方法能夠獨立於類產生變化【1】。如果能使用泛型方法解決問題，就不使用泛型類。MapTask類的runOldMapper方法就是一個泛型方法，其簽名如下：

private <INKEY,INVALUE,OUTKEY,OUTVALUE>
  void runOldMapper(final JobConf job,
                    final TaskSplitIndex splitIndex,
                    final TaskUmbilicalProtocol umbilical,
                    TaskReporter reporter
                    )

MapRunnable<INKEY,INVALUE,OUTKEY,OUTVALUE> runner = ReflectionUtils.newInstance(job.getMapRunnerClass(), job);

public void run(RecordReader<K1, V1> input, OutputCollector<K2, V2> output,
                  Reporter reporter)

public static <T> T newInstance(Class<T> theClass, Configuration conf) {
    T result;
    try {
      Constructor<T> meth = (Constructor<T>) CONSTRUCTOR_CACHE.get(theClass);
      if (meth == null) {
        meth = theClass.getDeclaredConstructor(EMPTY_ARRAY);
        meth.setAccessible(true);
        CONSTRUCTOR_CACHE.put(theClass, meth);
      }
      result = meth.newInstance();
    } catch (Exception e) {
      throw new RuntimeException(e);
    }
    setConf(result, conf);
    return result;
  }

該方法的作用是根據給定的型別和配置建立物件。在Java中，一個static方法無法訪問泛型類的泛型引數。因此，如果static方法需要使用泛型能力，就必須使其成為泛型方法【1】。靜態泛型方法經常用於一些工具類作為建立物件的工具。具體看方法實現，首先檢視快取中有沒有該型別的構造方法物件，這個快取物件是這樣實現的：

 /** 
   * Cache of constructors for each class. Pins the classes so they
   * can't be garbage collected until ReflectionUtils can be collected.
   */
  private static final Map<Class<?>, Constructor<?>> CONSTRUCTOR_CACHE = 
    new ConcurrentHashMap<Class<?>, Constructor<?>>();

public class Maps{
	public static <K,V> Map<K,V> map(){
		return new HashMap<K,V>();
	}
	public static <K,V> ConcurrentHashMap<K,V> cMap(){
		return new ConcurrentHashMap<K,V>();
	}
	
	public static void main(String[] args){
		Map<Class<?>, Constructor<?>> CONSTRUCTOR_CACHE = Maps.cMap();
	}
}

final Constructor<T> c = getConstructor0(empty, Member.DECLARED);
cachedConstructor = c;
Constructor<T> tmpConstructor = cachedConstructor;
return tmpConstructor.newInstance((Object[])null);

回到run方法，MapRunnable的run有兩種實現：一是普通的MapRunner，另一個是多執行緒MultithreadedMapRunner，這裡先學習前者。每個MapRunner物件中都有一個Mapper泛型物件用於執行使用者提交的Map函式。

this.mapper = ReflectionUtils.newInstance(job.getMapperClass(), job);
mapper.map(key, value, output, reporter);

前面的文章提到過，collect方法是由使用者的map函式呼叫的，例如Grep應用的mapper類RegexMap類中的map函式：

public void map(K key, Text value,
                  OutputCollector<Text, LongWritable> output,
                  Reporter reporter)
    throws IOException {
    String text = value.toString();
    Matcher matcher = pattern.matcher(text);
    while (matcher.find()) {
      output.collect(new Text(matcher.group(group)), new LongWritable(1));
    }

collector.collect(key, value, partitioner.getPartition(key, value, numPartitions));

private final ReentrantLock spillLock = new ReentrantLock();

Thinking in Java中對可重入互斥鎖的解釋為：ReentrantLock允許你嘗試著獲取但最終未獲取鎖，這樣如果其他人已經獲取了這個鎖，那你就可以決定離開去執行其他一些事情，而不是等待直至這個鎖被釋放。在Java中顯式使用鎖物件Lock的情況比較少，因為Lock物件必須顯式地建立、鎖定和釋放。但有時synchronized關鍵字不能實現一些特殊需求：嘗試著獲取鎖且最終獲取失敗，或者嘗試獲取鎖一段時間，然後放棄。這裡舉一個書上的例子：

private ReentrantLock lock = new ReentrantLock();
	public void untimed(){
		boolean captured = lock.tryLock();
		try{
			System.out.println("tryLock(): " + captured);
		}finally{
			if(captured)
				lock.unlock();
		}
	}

lock.tryLock()如果沒有獲得鎖，captured為false，此時不會阻塞執行緒，而是會繼續執行下面語句輸出一行。在finally塊中，根據是否捕獲到鎖來釋放鎖。另外，在ReentrantLock上阻塞的任務具備可以被中斷的能力，這與在synchronized方法或臨界區上阻塞的任務不同（後者是不可中斷的阻塞，不會丟擲InterruptedException異常）。在collect方法中之所以使用可重入鎖，我想就是因為使用上述後一種特性，使其在阻塞時可中斷，丟擲異常。在MapOutputBuffer中還定義了兩個條件變數spillReady和spillDone

private final Condition spillDone = spillLock.newCondition();
private final Condition spillReady = spillLock.newCondition();

spillReady.signal();

spillDone.await();

   public void run() {
        spillLock.lock();
        spillThreadRunning = true;
        try {
          while (true) {
            spillDone.signal();
            while (kvstart == kvend) {
              spillReady.await();
            }
            try {
              spillLock.unlock();
              sortAndSpill();
           }finally {
              spillLock.lock();
              ...
              kvstart = kvend;
              bufstart = bufend;
            }
          }
        } catch (InterruptedException e) {
          Thread.currentThread().interrupt();
        } finally {
          spillLock.unlock();
          spillThreadRunning = false;
        }
      }
    }

首先獲得鎖，然後喚醒在spillDone上等待的collect執行緒，表明spill溢寫結束，可以繼續寫入緩衝區了。當進入正常寫入狀態後（kvstart==kvend），呼叫spillReady.await()，掛起spill執行緒，暫停溢寫，直到collect執行緒再次呼叫startSpill方法。當spill執行緒被喚醒並再次獲得鎖時，呼叫sortAndSpill對緩衝區資料進行依次快速排序然後寫入磁碟。

注意，在呼叫await，signal和signalAll之前必須擁有鎖，這裡兩個Condition變數在使用前必須擁有鎖spillLock。collect同步控制的核心邏輯如下，用於對比：

     spillLock.lock();
      try {
        boolean kvfull;
        do {
          // sufficient acct space
          kvfull = kvnext == kvstart;
          final boolean kvsoftlimit = ((kvnext > kvend)
              ? kvnext - kvend > softRecordLimit
              : kvend - kvnext <= kvoffsets.length - softRecordLimit);
          if (kvstart == kvend && kvsoftlimit) {
            startSpill();
          }
          if (kvfull) {
            while (kvstart != kvend) {
              spillDone.await();
            }
          }
        } while (kvfull);
      } finally {
        spillLock.unlock();
      }

有一個疑問，如果spill執行緒發現collect正在寫緩衝區而掛起，那麼spill獲得的鎖就掛起了，collect就獲得不到鎖了，也就無法呼叫startSpill方法去喚醒掛起的spill執行緒，這樣豈不是死鎖了？

研究了相關資料，找到答案：await操作在呼叫時會先釋放鎖，然後掛起執行緒，並將執行緒加入一個等待佇列。在呼叫signal時會讓等待佇列中第一個執行緒重新獲得鎖，並繼續執行。這樣spill掛起時，spillLock被釋放掉，collect執行緒會持續獲得鎖，直到滿足spill條件，呼叫startSpill方法，喚醒掛起的spill執行緒，當collect釋放鎖時，spill執行緒會重新獲得spillLock，並繼續執行。

使用Condition物件的目的相當於在鎖上加上一個條件，實現更精細的同步控制。這裡在同一個鎖spillLock上使用了兩個條件，spillReady條件表示只有緩衝區滿足一定條件才能發生spill，讀取緩衝區（消費者行為）；spillDone條件表示只有滿足一定條件（在這裡只有緩衝區滿的時候collect執行緒才掛起，即便是spill正在進行，緩衝區依然可以寫入，讀寫不衝突，這體現了環形緩衝區的優勢）才能寫緩衝區（生產者行為）。

這個地方關於條件的鎖的理解有什麼錯誤或不足請不吝賜教。關於緩衝區的操作比較複雜，參考書目【2】第8章內容中作者針對各類情況給出了圖文描述，推薦閱讀。

看看sortAndSpill做了些什麼，首先對bufstart和bufend下標之間的資料排序：

sorter.sort(MapOutputBuffer.this, kvstart, endPosition, reporter);

 Reducer<K,V,K,V> combiner = ReflectionUtils.newInstance(combinerClass, job);
      try {
        CombineValuesIterator<K,V> values = 
          new CombineValuesIterator<K,V>(kvIter, comparator, keyClass,  valueClass, job, Reporter.NULL,  inputCounter);
        while (values.more()) {
          combiner.reduce(values.getKey(), values, combineCollector, Reporter.NULL);
          values.nextKey();
        }
      } finally {
        combiner.close();
      }

最後研究一下Hadoop中的序列化【3】。Hadoop重新定義了序列化的機制，原因是：在Java序列化的過程中，序列化輸出中儲存了大量的附加資訊，導致序列化結果膨脹，對於需要儲存和處理大規模資料的Hadoop來說，需要一個新的序列化機制。使用Java實現物件的序列化簡單概括為：

1. 實現Serializable介面。

2. 在某種OutputStream的基礎上建立ObjectOutputStream物件。

3. 呼叫writeObject方法進行序列化。

反序列化的過程類似，只需要使用對應的輸入流，並呼叫readObject即可。Hadoop對序列化過程的優化為：同一個類物件的序列化結果只輸出一份元資料；重用物件，在已有物件上進行反序列化操作。

具體的實現機制是：

可序列化的物件需要實現Writable介面，該介面含有兩個方法：write(DataOutput out) 和 readFields(DataInput in)。前者藉助Java的DataOutput類物件將Java原生型別以位元組形式寫入二進位制流，後者從二進位制流中讀取欄位。

Hadoop還提供了帶有比較功能的WritableComparable介面，具有高效比較能力的RawComparator介面。前者兼具比較和序列化的功能；後者可以比較流中讀取的未被反序列化為物件的記錄，節省了建立物件的開銷，十分高效。

Hadoop中的Text型別即為一種常見的鍵型別，其宣告如下：

public class Text extends BinaryComparable
    implements WritableComparable<BinaryComparable> {}

public class ObjectWritable implements Writable, Configurable {

  private Class declaredClass;//實際型別的類物件
  private Object instance;//需要序列化的物件
  private Configuration conf;

  public ObjectWritable() {}
  
  public ObjectWritable(Object instance) {
    set(instance);
  }

  public ObjectWritable(Class declaredClass, Object instance) {
    this.declaredClass = declaredClass;
    this.instance = instance;
  }
...
}

Hadoop還提供了簡單的序列化框架API。通過Serialization實現獲得一個Serializer物件，可將一個物件轉換為一個位元組流的實現例項。在collect過程中，會將key/value序列化到緩衝區中。這裡使用了Serializer：

 private final Serializer<K> keySerializer;
 private final Serializer<V> valSerializer;

static class WritableSerializer implements Serializer<Writable> {
    private DataOutputStream dataOut;
    public void open(OutputStream out) {
      if (out instanceof DataOutputStream) {
        dataOut = (DataOutputStream) out;
      } else {
        dataOut = new DataOutputStream(out);
      }
    }
    public void serialize(Writable w) throws IOException {
      w.write(dataOut);
    }
    public void close() throws IOException {
      dataOut.close();
    }
  }

    keySerializer.serialize(key);
    valSerializer.serialize(value);

Collect過程結束，以後可能會就某個話題有補充。下篇計劃學習Reduce任務的過程。

2014.04.01