HDFS寫文件的流程淺析
提到大數據，我們一定繞不開Hadoop，某種意義上Hadoop就代表了大數據這一領域。作為Hadoop的兩大核心之一的HDFS，隨着大數據的快速發展也越來越被更多的人認識和使用。今天我們就簡單剖析一下HDFS寫文件的流程。
如圖所示，HDFS寫文件具體流程如下：
1.客戶端端首先通過DistributedFileSystem對象的create方法創建一個FSDataOutputStream（輸出流）對象；
2.客戶端通過DistributedFileSystem對象向NameNode發起一次RPC調用，在HDFS的Namespace中創建一個文件條目；
3.通過FSDataOutputStream對象，向DataNode寫入數據，數據首先被寫入FSDataOutputStream對象內部的Buffer中，然后數據被分割成一個個Packet數據包；
4.以Packet最小單位，基於Socket連接發送到按特定算法選擇的HDFS集群中一組DataNode中的一個節點上，在這組DataNode組成的Pipeline（數據量管道）上依次傳輸Packet；
5.這組DataNode組成的Pipeline反方向上，發送ack，最終由Pipeline中第一個DataNode節點將Pipeline ack發送給客戶端；
6.完成向文件寫入數據，客戶端在FSDataOutputStream對象上調用close方法，關閉流；
7.調用DistributedFileSystem對象的complete方法，通知NameNode文件寫入成功。
下面以2個問題來細化上述寫文件的流程：
Q1：數據往哪寫？
對於一個上千節點Hadoop集群，在寫文件是應該往哪幾個DataNode里面寫呢？這就涉及到HDFS副本放置策略相關內容了。HDFS采用多副本技術保證數據的高可用，在實現副本放置位置的選取過程中，需要考慮2方面因素：可靠性與網絡帶寬。以3副本為例，有如下情形：
情形1：3副本放置在同一台機器上，此種情形可靠性差，如果這台機器宕機，則數據所有副本都不可用；但由於所有副本在同一台機器上，寫入數據時都是在機器內部進行傳輸所以網絡帶寬開銷最小。
情形2：數據副本全部分散到不同的機架中。此種情形數據可靠性最高，即便是存儲數據副本的兩個機架都出現故障，還是有一個副本可用；但由於數據副本分散到不同的機架中，導致數據的讀寫需要跨交換機，網絡帶寬的開銷較大。
情形3：將情形1與情形2進行折中，得到如下存儲模式：
1.對於副本1，如果寫請求方所在機器是其中一個DataNode,則直接存放在本地,否則隨機在集群中選擇一個DataNode；
2.對於副本2，第二個副本存放於不同第一個副本的所在的機架；.
3.對於副本3，第三個副本存放於第二個副本所在的機架,但是屬於不同的節點。
這樣即便是有一個存儲數據副本的機架故障，數據依然可用；同時第二副本和第三副本在同一個機架中，數據的讀寫較情形2需要的網絡帶寬更小。
目前HDFS中副本放置策略的實現主要有兩種方式：
BlockPlacementPolicyDefault,
BlockPlacementPolicyWithNodeGroup。
集群默認采用BlockPlacementPolicyDefault方式，這里以此種實現為例。
分析代碼可知，選擇策略的核心方法為chooseTargets，具體流程如下：
1.如果發生寫請求的客戶端不是集群內部的DataNode節點，則在集群內隨機分配一個節點作為首節點
2.如果發生寫請求的客戶端本身就是集群中的一個DataNode，則第一個副本就放置在本節點上，這樣寫數據就是本地寫操作，避免了網絡帶寬的消耗；如果本地節點不滿足放置該副本的條件，則退而求其次，選擇一個與本地同機架的DataNode節點，如果同機架上的所有節點都不滿足條件，則在集群內隨機分配一個節點作為首節點
3.如果已經成功找到首節點，則在與首節點不同的機架內尋找第二個節點。如果找到符合要求的節點，則第二個節點分配完畢；如果在遠程機架內沒有找到合適的節點，則在首節點同機架內尋找，如果找到則第二節點分配完畢。
4.如果前兩個節點分配完畢，則進行第三個節點的分配。首選判斷前兩個節點是否在同一個機架內，如果是，則在遠程機架中尋找滿足條件的節點，如果找到，則第三個節點分配完畢，如果未找到；如果前兩個節點不在同一個機架內，則在第二個節點的機架內尋找，如果找到滿足條件的節點，則分配完畢，如果未找到，則在集群內隨機分配一個節點作為第三節點。
5.對於副本數大於3的情況，剩余節點則在集群內隨機選取。
當NameNode選擇出3個DataNode節點以后會將他們組成一個Pipeline返回給客戶端用以數據寫入，那Pipeline是怎樣構建的呢？其中的根據又是什么？這就涉及到HDFS中的另一塊內容——機架感知。在NameNode內部，將集群中所有的DataNode以一個樹的形式組織起來，如下圖所示，DataNode節點之間可以用距離來表示兩個節點的遠近程度，每對DataNode節點間的距離計算方式是通過尋找最近的公共祖先所需要的距離作為最終的結果，是一個典型的LCA最近公共祖先算法實現。
確定了節點見的距離的定義之后，再來看Pipeline的構建，其構建目標是使得Pipeline節點間距離之和最小。假設我們有一個客戶端和選取出的3個Nodes：DataNode1、DataNode2、DataNode3，則Pipeline生成過程如下：
（1）從DataNode1、DataNode2、DataNode3中找出與客戶端之間距離最短的節點，假設為DataNode2；
（2）從DataNode1、DataNode3中找出與DataNode2之間距離最短的節點，假設為DataNode3；
（3）從DataNode1中找出與DataNode4之間距離最短的節點，為DataNode1；
最后我們即得到一個“節點距離之和最小”的管道：客戶端、DataNode2、DataNode3、DataNode1。這其實是一個經典的圖論算法——TSP旅商算法的簡單實現。到此為止，客戶端獲得了一個由3個節點組成的Pipeline，就可以開始往里面寫入數據了。
Q2：數據怎樣寫
我們都知道HDFS中的文件是以Block塊的形式進行存儲的，那在寫入過程中是否也是以Block為單位進行的呢？其實不然，通過分析DFSOutputStream代碼我們可以發現，客戶端會先將數據寫到輸出流內部的緩沖區中，然后見數據組裝成一個個大小為64KB的Packet（數據包），而每個Packet又是由多個Chunk組成，每個Chunk由512字節數據和4字節校驗位組成。下圖所示為一個Packet的幀結構。
當寫入的數據達到一個Packet長度時，DFSOutputStream就會構造一個Packet並放入一個隊列dataQueue中，同時輸出流的內部線程DataStreamer會不斷從dataQueue中取出Packet發送到Pipeline中第一個DataNode上，並將該Packet移至另一個隊列ackQueue中。
輸出流的另一個內部線程ResponseProcessor則接收來自下游節點的ack信息，如果是一個成功的ack，則表示Pipeline中的節點都接收到了次Packet，ResponseProcessor就會將改Packet從ackQueue中刪除。在發送過程中，如果發生錯誤，所有未完成的Packet都會從ackQueue隊列中移除掉，然后重新創建一個新的Pipeline，排除掉出錯的那些DataNode節點，接着DataStreamer線程繼續從dataQueue隊列中發送Packet。寫數據的示意圖如下：
在DataNode端同樣以Packet為單位進行寫入與轉發。DataNode的BlockReceiver類負責從數據流管道中的上游節點接收Packet，然后保存到當前數據節點中，並將Packet轉發至數據流管道的下游節點。同時BlockReceiver還會啟動一個PacketResponder內部線程用於處理響應信息，該線程會接收來自下游節點的Ack響應，然后在響應中加入本機節點的狀態信息並將Acl轉發至上游節點。
本文簡單剖析了HDFS寫文件的流程，其中很多細節部分沒有詳細展開，HDFS寫文件的容錯機制也沒有介紹，有興趣的同學可以結合源碼進一步學習。