Hadoop源代碼分析【1-5】

寫在前面： 博主是一名大資料的初學者，昵稱來源于《愛麗絲夢游仙境》中的Alice和自己的昵稱，作為一名互聯網小白，寫博客一方面是為了記錄自己的學習歷程，一方面是希望能夠幫助到很多和自己一樣處于起步階段的萌新，由于水平有限，博客中難免會有一些錯誤，有紕漏之處懇請各位大佬不吝賜教！個人小站:http://alices.ibilibili.xyz/ , 博客主頁:https://alice.blog.csdn.net/
盡管當前水平可能不及各位大佬，但我還是希望自己能夠做得更好，因為一天的生活就是一生的縮影，我希望在最美的年華，做最好的自己！

Hadoop源代碼分析（一）

Google 的核心競爭技術是它的計算平臺，Google 的大牛們用了下面 5 篇文章，介紹了它們的計算設施，

GoogleCluster： http://research.google.com/archive/googlecluster.html

Chubby：http://labs.google.com/papers/chubby.html

GFS：http://labs.google.com/papers/gfs.html

BigTable：http://labs.google.com/papers/bigtable.html

MapReduce：http://labs.google.com/papers/mapreduce.html

很快，Apache 上就出現了一個類似的解決方案，目前它們都屬于 Apache 的 Hadoop 專案，對應的分別是：

Chubby-->ZooKeeper 
GFS-->HDFS 
BigTable-->HBase 
MapReduce-->Hadoop

目前，基于類似思想的 Open Source 專案還很多，如 Facebook 用于用戶分析的 Hive，

HDFS 作為一個分布式檔案系統，是所有這些專案的基礎，分析好 HDFS，有利于了解其他系統，由于 Hadoop 的 HDFS 和 MapReduce 是同一個專案，我們就把他們放在一塊，進行分析，

下圖是 MapReduce 整個專案的頂局包圖和他們的依賴關系，Hadoop 包之間的依賴關系比較復雜，原因是 HDFS 提供了一個分布式檔案系統，該系統提供 API，可以屏蔽本地檔案系統和分布式檔案系統，甚至象 Amazon S3 返樣的在線存盤系統，這就造成了分布式檔案系統的實作，或者是分布式檔案系統的底層的實作，依賴于某些貌似高層的功能，功能的相互參考，造成了蜘蛛網型的依賴關系，一個典型的例子就是包 conf，conf 用于讀取系統配置，它依賴于 fs，主要是讀取組態檔的時候，需要使用檔案系統，而部分的檔案系統的功能，在包 fs 中被抽象了，

Hadoop 的關鍵部分集中于圖中藍色部分，這也是我們考察的重點，

Hadoop源代碼分析（二）

下面給出了 Hadoop 的包的功能分析，

Hadoop源代碼分析（三）

由于 Hadoop 的 MapReduce 和 HDFS 都有通信的需求，需要對通信的物件進行序列化，Hadoop 并沒有采用 Java 的序列化，而是引入了它自己的系統，

org.apache.hadoop.io中定義了大量的可序列化物件，他們都實作了 Writable 介面，實作了 Writable 介面的一個典型例子如下：

 public class MyWritable implements Writable { 
 // Some data 
 private int counter; 
 private long timestamp; 
 
 public void write(DataOutput out) throws IOException { 
 out.writeInt(counter); 
 out.writeLong(timestamp); 
 } 
 
 public void readFields(DataInput in) throws IOException { 
 counter = in.readInt(); 
 timestamp = in.readLong(); 
 } 
 

 public static MyWritable read(DataInput in) throws IOException { 
 MyWritable w = new MyWritable(); 
 w.readFields(in); 
 return w; 
 } 
 }

其中的 write 和 readFields 分別實作了把物件序列化和反序列化的功能，是 Writable 介面定義的兩個方法，下圖給出了龐大的 org.apache.hadoop.io 中物件的關系，

這里，把 ObjectWritable 標為紅色，是因為相對于其他物件，它有不同的地位，當我們討論 Hadoop的 RPC時，我們會提到RPC上交換的資訊， 必須是 Java 的基本型別， String 和 Writable 介面的實作類， 以及元素為以上型別的陣列，

ObjectWritable物件保存了一個可以在 RPC上傳輸的物件和物件的型別資訊， 這樣，我們就有了一個萬能的， 可以用于客戶端 / 服務器間傳輸的
Writable 物件，例如，我們要把上面例子中的物件作為 RPC請求，需要根據 MyWritable 創建一個 ObjectWritable ，ObjectWritable 往流里會寫如下資訊，

物件類名長度，物件類名，物件自己的串行化結果

這樣，到了對端， ObjectWritable 可以根據物件類名創建對應的物件，并解串行，應該注意到， ObjectWritable 依賴于 WritableFactories ，那里存盤了 Writable 子類對應的工廠，我們需要把 MyWritable 的工廠，保存在 WritableFactories 中（通過 WritableFactories. setFactory ），

Hadoop源代碼分析（四）

介紹完 org.apache.hadoop.io 以后，我們開始來分析 org.apache.hadoop.rpc ，RPC采用客戶機 / 服務器模式， 請求程式就是一個客戶機，而服務提供程式就是一個服務器，當我們討論 HDFS的，通信可能發生在：

• Client-NameNode 之間，其中 NameNode 是服務器
• Client-DataNode 之間，其中 DataNode 是服務器
• DataNode-NameNode 之間，其中 NameNode 是服務器
• DataNode-DateNode 之間，其中某一個 DateNode 是服務器，另一個是客戶端

如果我們考慮 Hadoop的 Map/Reduce以后，這些系統間的通信就更復雜了，為了解決這些客戶機 / 服務器之間的通信， Hadoop引入了一個 RPC框架，該 RPC框架利用的 Java 的反射能力，避免了某些 RPC解決方案中需要根據某種介面語言（如 CORBA的
IDL）生成存根和框架的問題，但是，該 RPC框架要求呼叫的引數和回傳結果必須是 Java 的基本型別， String 和 Writable 介面的實作類，以及元素為以上型別的陣列，同時，介面方法應該只拋出 IOException 例外，

既然是 RPC，當然就有客戶端和服務器，當然， org.apache.hadoop.rpc也就有了類 Client 和類 Server ，但是類 Server 是一個抽象類，類 RPC封裝了 Server ，利用反射，把某個物件的方法開放出來，變成 RPC中的服務器，

下圖是 org.apache.hadoop.rpc 的類圖，

Hadoop源代碼分析（五）

既然是 RPC，自然就有客戶端和服務器，當然， org.apache.hadoop.rpc 也就有了類 Client 和類 Server ，在這里我們來仔細考察 org.apache.hadoop.rpc.Client ，下面的圖包含了 org.apache.hadoop.rpc.Client 中的關鍵類和關鍵方法，

由于 Client 可能和多個 Server 通信，典型的一次 HDFS讀，需要和 NameNode打交道，也需要和某個 / 某些 DataNode通信，這就意味著某一個 Client 需要維護多個連接，同時，為了減少不必要的連接，現在 Client 的做法是拿 ConnectionId （圖中最右側）來做為 Connection 的 ID，ConnectionId 包括一個 InetSocketAddress （IP 地址+埠號或主機名 +埠號）物件和一個用戶資訊物件，這就是說，同一個用戶到同一個 InetSocketAddress 的通信將共享同一個連接，

連接被封裝在類 Client.Connection 中，所有的 RPC呼叫，都是通過 Connection ，進行通信，一個 RPC呼叫，自然有輸入引數，輸出引數和可能的例外，同時，為了區分在同一個 Connection 上的不同呼叫，每個呼叫都有唯一的 id ，呼叫是否結束也需要一個標記，所有的這些都體現在物件 Client.Call 中，Connection 物件通過一個 Hash表，維護在這個連接上的所有 Call ：

private Hashtable<Integer ,Call>calls = new Hashtable<Integer,Call>();

一個 RPC呼叫通過 addCall ，把請求加到 Connection 里，為了能夠在這個框架上傳輸 Java 的基本型別， String 和 Writable 介面的實作類，以及元素為以上型別的陣列，我們一般把 Call 需要的引數打包成為 ObjectWritable 物件，

Client.Connection 會通過 socket 連接服務器，連接成功后回校驗客戶端 / 服務器的版本號（Client.ConnectionwriteHeader()方法），校驗成功后就可以通過 Writable 物件來進行請求的發送 / 應答了，注意，每個 Client.Connection 會起一個執行緒，不斷去讀取 socket ，并將收到的結果解包，找出對應的 Call ，設定 Call 并通知結果已經獲取，

Call 使用 Obejct 的 wait 和 notify ，把 RPC上的異步訊息互動轉成同步呼叫，

還有一點需要注意，一個 Client 會有多個 Client.Connection ，這是一個很自然的結果，

小結

Hadoop源代碼分析【1-5】主要為大家科普了Hadoop下的各種包的功能分析，以及Hadoop下兩大核心HDFS和MapReduce如何基于RPC框架去實作通信，資料傳輸，

本篇文章就到這里，更多精彩文章及福利，敬請關注博主原創公眾號【猿人菌】!