1、HDFS的運行機制主要內(nèi)容HDFS中數(shù)據(jù)流的讀寫HDFS的HA機制HDFS的Federation機制主要內(nèi)容HDFS中數(shù)據(jù)流的讀寫HDFS的HA機制HDFS的Federation機制HDFS中數(shù)據(jù)流的讀寫包括以下幾個內(nèi)容：RPC實現(xiàn)流程RPC實現(xiàn)模型文件的讀取文件的寫入文件的一致模型什么是RPC？RPC（Remote Procedure Call）遠程過程調(diào)用，是一種協(xié)議，它是一種通過網(wǎng)絡從遠程計算機程序上請求服務，而不需要了解底層網(wǎng)絡技術(shù)的協(xié)議。RPC協(xié)議假定某些傳輸協(xié)議的存在，如TCP或UDP，為通信程序之間攜帶信息數(shù)據(jù)。在OSI網(wǎng)絡通信模型中，RPC跨越了傳輸層和應用層。RPC使得開發(fā)包

2、括網(wǎng)絡分布式多程序在內(nèi)的應用程序更加容易。RPC采用客戶機/服務器模式。請求程序就是一個客戶機，而服務提供程序就是一個服務器。hadoop的整個體系結(jié)構(gòu)就是構(gòu)建在RPC之上的，Hadoop在其內(nèi)部實現(xiàn)了一個基于IPC模型的RPC(見org.apache.hadoop.ipc)。因為hadoop內(nèi)部采用了master/slave架構(gòu)，那么其內(nèi)部通信和與客戶端的交互就是必不可少的了。RPC實現(xiàn)流程一個典型的RPC框架主要包括以下幾個部分：通信模塊：兩個相互協(xié)作的通信模塊實現(xiàn)請求-應答協(xié)議。代理程序：客戶端和服務器端均包含代理程序。調(diào)度程序：調(diào)度程序接受來自通信模塊的請求消息，并根據(jù)其中的標志選擇一

3、個代理程序處理。RPC實現(xiàn)流程一個RPC請求從發(fā)送到獲取處理結(jié)果，所經(jīng)歷的步驟如下：1、客戶程序以本地方式調(diào)用系統(tǒng)產(chǎn)生的Stub程序；2、該Stub程序?qū)⒑瘮?shù)調(diào)用信息按照網(wǎng)絡通信模塊的要求封裝成消息包，并交給通信模塊發(fā)送到遠程服務器端；3、遠程服務器端接收到此消息后，將此消息發(fā)送給相應的Stub程序；4、Stub程序拆封消息，形成被調(diào)過程要求的形式，并調(diào)用對應的函數(shù)；5、被調(diào)用函數(shù)按照所獲參數(shù)執(zhí)行，并將結(jié)果返回給Stub程序；6、Stub將此結(jié)果封裝成消息，通過網(wǎng)絡通信模塊逐級地傳送給客戶程序；Hadoop RPC基本框架Hadoop RPC主要對外提供兩種接口：public static V

4、ersionedProtocol getProxy/waitForProxy(): 構(gòu)造一個客戶端代理對象（該對象實現(xiàn)了某種協(xié)議），用于向服務器端發(fā)送RPC請求；public static Server getServer(): 為某個協(xié)議(實際上是Java接口)實例構(gòu)造 一個服務器對象，用于處理客戶端發(fā)送的請求；Hadoop RPC使用方法：1、定義RPC協(xié)議。RPC協(xié)議是客戶端和服務器端之間的通信接口，他定義了服務器端對外提供的服務接口。如以下代碼所示，我們定義了一個ClientProtocol通信接口，他聲明了兩個方法：echo()和add()需要注意的是，hadoop中所有自定義RPC

5、接口都需要繼承VersionedProtocol 接口，他描述了協(xié)議的版本信息。Hadoop RPC基本框架interface ClientProtocol extends org.apach.hadoop.ipc.VersionedProtocol /版本號。默認情況下，不同版本號的RPC Client和Server之間不能相互通信public static final long versionID = 1L;String echo(String value) throws IOException;int add(int v1,int v2) throws IOException; Hado

6、op RPC基本框架2、實現(xiàn)RPC協(xié)議。Hadoop RPC協(xié)議通常是一個Java接口，用戶需要實現(xiàn)接口，如以下代碼所示，對ClientProtocol接口進行簡單的實現(xiàn)：public static class ClientProtocolImpl implements ClientProtocol public long getProtocolVersion(String protocol , long clientVersion) return ClientProtocol.versionID; public String echo(String value) throws IOExcep

7、tion return value; public int add(int v1,int v2) throws IOException return v1+v2; Hadoop RPC基本框架3、構(gòu)造并啟動RPC Server。直接使用new RPC.Builder(conf)構(gòu)造一個RPC Server，并調(diào)用函數(shù)start()啟動該Server：public class MyServer public static final String ADDRESS=localhost;public static final int PORT = 2454; public static void m

8、ain(String args)throws Exception final Server server = new RPC.Builder(new Configuration().setProtocol(ClientProtocol.class) .setInstance(new ClientProtocolImpl().setBindAddress(ADDRESS).setPort(MyServer.PORT) .setNumHandlers(5).build();server.start();其中，setBindAddress和setPort分別表示服務器的host和監(jiān)聽端口號，而set

9、NumHandlers表示服務器端處理請求的線程數(shù)目，到此為止，服務器處理監(jiān)聽狀態(tài)，等待客戶端請求到達。Hadoop RPC基本框架4、構(gòu)造RPC Client，并發(fā)送RPC請求。使用靜態(tài)方法getProxy()構(gòu)造客戶端代理對象，直接通過代理對象調(diào)用遠程端的方法，具體如下所示：public class MyClient public static void main(String args) throws Exception ClientProtocol proxy = (ClientProtocol) RPC.getProxy(ClientProtocol.class, 1L, new I

10、netSocketAddress(MyServer.ADDRESS, MyServer.PORT), new Configuration();final int result = proxy.add(3, 5);String r = proxy.echo(result+);System.out.println(r);經(jīng)過以上四步，我們便利用Hadoop RPC搭建了一個非常高效的客戶機/服務器網(wǎng)絡模型。RPC ServerHadoop RPC的實現(xiàn)主要在org.apache.hadoop.ipcServer：RPC Server實現(xiàn)了一種抽象的RPC服務，同時提供Call隊列。RPC Serv

11、er結(jié)構(gòu):Server.Listener：RPC Server的監(jiān)聽者，用來接收RPC Client的連接請求和數(shù)據(jù)，其中將數(shù)據(jù)封裝成CALL后PUSH到CALL隊列中。Server.Handler：RPC Server的CALL處理者，和Server.Listener通過CALL隊列交互。Server.Responder：RPC Server響應者，Server.Handler按照異步非阻塞的方式向RPC Client發(fā)送響應，如果有未發(fā)送出去的數(shù)據(jù)，交由Server.Responder來處理完成。Server.Connection：RPC Server數(shù)據(jù)的接收者。提供接收數(shù)據(jù)，解析數(shù)據(jù)包

12、的功能。Server.Call：持有客戶端的Call信息。RPC Server的主要流程RPC Server作為服務的提供者主要有兩部分組成：接收Call調(diào)用和處理Call調(diào)用。接收Call調(diào)用負責接收來自RPC Client的調(diào)用請求，編碼成Call對象放入到Call隊列中，這一過程有Server.Listener完成。具體步驟如下：1.Listener線程監(jiān)聽RPC Client發(fā)過來的數(shù)據(jù)2.當有數(shù)據(jù)可以接收時，調(diào)用Connection的readAndProcess方法3.Connection邊接受數(shù)據(jù)邊處理數(shù)據(jù)，當接到一個完整的Call包，則構(gòu)建一個Call對象，PUSH到Call 隊

13、列中，有Handler來處理Call隊列中的所有Call處理完的Call調(diào)用負責處理Call隊列中的每一個調(diào)用請求，由Handler線程來完成。4.Handler線程監(jiān)聽Call隊列，如果Call隊列非空，按FIFO規(guī)則從Call隊列中取出Call5.將Call交給RPC.Server來處理6.借助JDK提供的Method，完成對目標方法的調(diào)用，目標方法由具體的業(yè)務邏輯實現(xiàn)7.返回響應。Server.Handler按照異步非阻塞的方式向RPC Client發(fā)送響應，如果有未發(fā)送出的數(shù)據(jù)，則交由Server.Responder來完成。完整的交互過程如下圖所示：RPC ClientRPC Clie

14、nt 結(jié)構(gòu)：Client.ConnectionId：到RPC Server對象連接的標識。Client.Call：Call調(diào)用信息。Client.ParallelResults：Call響應。RPC.Invoker：對InvocationHandler的實現(xiàn)，提供invoke方法，實現(xiàn)RPC Client對RPC Server對象的調(diào)用。RPC.Invocation：用來序列化和反序列化RPC Client的調(diào)用信息。（主要應用JAVA的反射機制和InputStream/OutputStream）RPC ClientRPC Client主要流程1. RPC Client發(fā)起RPC Call，通

15、過JAVA反射機制 轉(zhuǎn)化為對Client.call調(diào)用2.調(diào)用getConnection得到與RPC Server的鏈接， 每一個RPC Client都維護一個HashMap結(jié)構(gòu)的 到RPC Server的連接池。如圖所示：3.通過Connection將序列化后的參數(shù)發(fā)送到RPC 服務端4.阻塞方式等待RPC服務端返回響應。RPC實現(xiàn)模型 需要詳細說的是RPC在服務端的模型，它由一系列實體組成，分別負責調(diào)用的整個流程，如圖所示：實體介紹Listener監(jiān)聽RPC server的端口，如果客戶端有連接請求到達，它就接受連接，然后把連接轉(zhuǎn)發(fā)到某個Reader，讓Reader去讀取那個連接的數(shù)據(jù)。如

16、果有多個 Reader的話，當有新連接過來時，就在這些Reader間順序分發(fā)。Reader Reader的職責就是從某個客戶端連接中讀取數(shù)據(jù)流，然后把它轉(zhuǎn)化成調(diào)用對象(Call)，然后放到調(diào)用隊列(call queue)里實體介紹Handler 真正做事的實體。它從調(diào)用隊列中獲取調(diào)用信息，然后反射調(diào)用真正的對象，得到結(jié)果，然后再把此次調(diào)用放到響應隊列(response queue)里Responder 它不斷地檢查響應隊列中是否有調(diào)用信息，如果有的話，就把調(diào)用的結(jié)果返回給客戶端。文件讀取流程1、使用HDFS提供的客戶端Client，向遠程的Namenode發(fā)起RPC請求；2、Namenode會

17、視情況返回文件的部分或者全部block列表，對于每個block，Namenode都會返回有該block拷貝的DataNode地址；3、客戶端Client會選取離客戶端最近的DataNode來讀取block；如果客戶端本身就是DataNode，那么將從本地直接獲取數(shù)據(jù)；4、讀取完當前block的數(shù)據(jù)后，關閉當前的DataNode鏈接，并為讀取下一個block尋找最佳的DataNode；5、當讀完列表block后，且文件讀取還沒有結(jié)束，客戶端會繼續(xù)向Namenode獲取下一批的block列表；6、讀取完一個block都會進行checksum驗證，如果讀取datanode時出現(xiàn)錯誤，客戶端會通知Na

18、menode，然后再從下一個擁有該block拷貝的datanode繼續(xù)讀。文件的讀取客戶端及讀取HDFS中的數(shù)據(jù)的流程圖，如下圖所示：文件寫入流程1、使用HDFS提供的客戶端Client，向遠程的Namenode發(fā)起RPC請求2、Namenode會檢查要創(chuàng)建的文件是否已經(jīng)存在，創(chuàng)建者是否有權(quán)限進行操作，成功則會為文件創(chuàng)建一個記錄，否則會讓客戶端拋出異常；3、當客戶端開始寫入文件的時候，客戶端會將文件切分成多個packets，并在內(nèi)部以數(shù)據(jù)隊列“data queue（數(shù)據(jù)隊列）”的形式管理這些packets，并向Namenode申請blocks，獲取用來存儲replicas的合適的datanod

19、e列表，列表的大小根據(jù)Namenode中replication的設定而定；4、開始以pipeline（管道）的形式將packet寫入所有的replicas中。開發(fā)庫把packet以流的方式寫入第一個datanode，該datanode把該packet存儲之后，再將其傳遞給在此pipeline中的下一個datanode，直到最后一個datanode，這種寫數(shù)據(jù)的方式呈流水線的形式。文件寫入流程5、最后一個datanode成功存儲之后會返回一個ack packet（確認隊列），在pipeline里傳遞至客戶端，在客戶端的開發(fā)庫內(nèi)部維護著ack queue，成功收到datanode返回的ack pa

20、cket后會從ack queue移除相應的packet。6、如果傳輸過程中，有某個datanode出現(xiàn)了故障，那么當前的pipeline會被關閉，出現(xiàn)故障的datanode會從當前的pipeline中移除，剩余的block會繼續(xù)剩下的datanode中繼續(xù)以pipeline的形式傳輸，同時Namenode會分配一個新的datanode，保持replicas設定的數(shù)量。7、客戶端完成數(shù)據(jù)的寫入后，會對數(shù)據(jù)流調(diào)用close()方法，關閉數(shù)據(jù)流；8、只要寫入了dfsreplicationmin的復本數(shù)（默認為1），寫操作就會成功，并且這個塊可以在集群中異步復制，直到達到其目標復本數(shù)（dfsrepli

21、cation的默認值為3），因為namenode已經(jīng)知道文件由哪些塊組成，所以它在返回成功前只需要等待數(shù)據(jù)塊進行最小量的復制。文件的寫入客戶端將數(shù)據(jù)寫入HDFS的流程圖，如下圖所示： 文件的一致模型文件系統(tǒng)的一致模型(coherency model)描述了對文件讀寫的數(shù)據(jù)可見性，HDFS為性能犧牲了一些POSIX要求，因此一些操作與你期望的可能不同。在創(chuàng)建一個文件之后，希望它能在文件系統(tǒng)的命名空間中立即可見，例如：path p = new path(“p”); Fs.create(p); assertThat(fs.exists(p), is(true);文件的一致模型但是，寫入文件的內(nèi)容并不

22、保證能立即可見，即使數(shù)據(jù)流已經(jīng)刷新并存儲。所以文件長度顯示為0：path p = new path(“p”);OutputStream out = fs.create(p);Out.write(“content”.getBytes(“UTF - 8”);Out.flush();assertThat(fs.getfIleStatus(p).getLen(), is(0L);一旦寫入的數(shù)據(jù)超過一個塊的數(shù)據(jù)，新的讀取者就能看見第一個塊。對于之后的塊也是這樣。總之，它始終是當前正在被寫入的塊時，其他讀取者是看不見它的。文件的一致模型HDFS提供一個方法來強制所有的緩存與數(shù)據(jù)節(jié)點同步， 即對FSData

23、OutputStream調(diào)用sync()方法。當sync()方法返回成功后，對所有新的reader而言，HDFS能保證文件中到目前為止寫入的數(shù)據(jù)均可見且一致：Path p = new path(“p”);FSDataOutputStream out = fs.create(p);Out.write(“content”.getBytes(“UTF - 8”);Out.flush();Out.sync();assertThat(fs.getFileStatus(p).getLen(),is(long)”content”.length();文件的一致模型該類操作類似于Unix中的fsync系統(tǒng)調(diào)用為

24、一個文件描述符提交緩沖數(shù)據(jù)，例如：利用Java API寫入一個本地文件，我們肯定能夠看到刷新流和同步之后的內(nèi)容。在HDFS中關閉一個文件其實還執(zhí)行了一個隱含的sync()。應用設計的重要性這個一致模型與具體設計應用程序的方法有關。如果不調(diào)用sync()，那么一旦客戶端或系統(tǒng)發(fā)生故障，就可能失去一個塊的數(shù)據(jù)。對很多應用來說，這是不可接受的，所以我們應該在適當?shù)牡胤秸{(diào)用sync()，例如在寫入一定的記錄或字節(jié)之后。盡管sync()操作被設計為盡量減少HDFS負載，但它仍然有開銷，所以在數(shù)據(jù)健壯性和吞吐量之間就會有所取舍。應用依賴就比較能接受，通過不同的sync()頻率來衡量應用程序，最終找到一個合

25、適的平衡。主要內(nèi)容HDFS中數(shù)據(jù)流的讀寫HDFS的HA機制HDFS的Federation機制HDFS的HA（High Availability）機制為什么有HA機制？在Hadoop 2.0之前,在HDFS 集群中NameNode 存在單點故障 (SPOF)。對于只有一個NameNode 的集群，如果NameNode 機器出現(xiàn)故障(比如宕機或是軟件、硬件升級)，那么整個集群將無法使用，直到NameNode 重新啟動。那么如何來解決這個問題呢？HDFS的HA機制HDFS 的HA 功能通過配置Active/Standby 兩個NameNodes 實現(xiàn)在集群中對NameNode 的熱備來解決上述問題。

26、如果出現(xiàn)故障，如機器崩潰或機器需要升級維護，這時可通過此種方式將NameNode 很快的切換到另外一臺機器。HA的集群架構(gòu)1、在一個典型的HDFS(HA) 集群中，使用兩臺單獨的機器配置為NameNodes 。在任何時間點，確保NameNodes 中只有一個處于Active 狀態(tài)，其他的處在Standby 狀態(tài)。其中ActiveNameNode 負責集群中的所有客戶端操作，StandbyNameNode 僅僅充當備機，保證一旦ActiveNameNode 出現(xiàn)問題能夠快速切換。HA的集群架構(gòu)2、為了能夠?qū)崟r同步Active和Standby兩個NameNode的元數(shù)據(jù)信息（實際上editlog）

27、，需提供一個共享存儲系統(tǒng)，可以是NFS、QJM（Quorum Journal Manager）或者Bookeeper，Active Namenode將數(shù)據(jù)寫入共享存儲系統(tǒng)，而Standby監(jiān)聽該系統(tǒng)，一旦發(fā)現(xiàn)有新數(shù)據(jù)寫入，則讀取這些數(shù)據(jù)，并加載到自己內(nèi)存中，以保證自己內(nèi)存狀態(tài)與Active NameNode保持基本一致，如此這般，在緊急情況下standby便可快速切為active namenode。3、為了實現(xiàn)快速切換，Standby 節(jié)點獲取集群的最新文件塊信息也是很有必要的。為了實現(xiàn)這一目標，DataNode 需要配置NameNodes 的位置，并同時給他們發(fā)送文件塊信息以及心跳檢測。HA

28、架構(gòu)圖HA集群架構(gòu)注意：Secondary NameNode。它不是HA，它只是階段性的合并edits和fsimage，以縮短集群啟動的時間。當NameNode失效的時候，Secondary NN并無法立刻提供服務，Secondary NN甚至無法保證數(shù)據(jù)完整性：如果NN數(shù)據(jù)丟失的話，在上一次合并后的文件系統(tǒng)的改動會丟失。主要內(nèi)容HDFS中數(shù)據(jù)流的讀寫HDFS的HA機制HDFS的Federation機制HDFS的Federation機制為什么要有Federation機制呢？ 前面說了在Hadoop 2.0之前，HDFS的單NameNode設計帶來諸 多問題，包括單點故障、內(nèi)存受限，制約集群擴展

29、性和缺乏隔離機制（不同業(yè)務使用同一個NameNode導致業(yè)務相互影響）等，為了解決這些問題，除了用基于共享存儲的HA解決方案我們還可以用HDFS的Federation機制來解決這個問題。HDFS的Federation機制什么是Federation機制？HDFS Federation是指HDFS集群可同時存在多個NameNode，這些NameNode分別管理一部分數(shù)據(jù)，且共享所有DataNode的存儲資源。這種設計可解決單NameNode存在的以下幾個問題：（1）HDFS集群擴展性。多個NameNode分管一部分目錄，使得一個集群可以擴展到更多節(jié)點，不再像1.0中那樣由于內(nèi)存的限制制約文件存儲數(shù)

30、目。HDFS的Federation機制（2）性能更高效。多個NameNode管理不同的數(shù)據(jù)，且同時對外提供服務，將為用戶提供更高的讀寫吞吐率。（3）良好的隔離性。用戶可根據(jù)需要將不同業(yè)務數(shù)據(jù)交由不同NameNode管理，這樣不同業(yè)務之間影響很小。需要注意的，HDFS Federation并不能解決單點故障問題，也就是說，每個NameNode都存在在單點故障問題，你需要為每個namenode部署一個backup namenode以應對NameNode掛掉對業(yè)務產(chǎn)生的影響。Federation架構(gòu)Federation架構(gòu)圖，如下所示：Federation架構(gòu)1、為了水平擴展namenode，fed

31、eration使用了多個獨立的namenode/namespace。這些namenode之間是聯(lián)合的，也就是說，他們之間相互獨立且不需要互相協(xié)調(diào)，各自分工，管理自己的區(qū)域。分布式的datanode被用作通用的數(shù)據(jù)塊存儲存儲設備。每個datanode要向集群中所有的namenode注冊，且周期性地向所有namenode發(fā)送心跳和塊報告，并執(zhí)行來自所有namenode的命令。2、一個block pool由屬于同一個namespace的數(shù)據(jù)塊組成，每個datanode可能會存儲集群中所有block pool的數(shù)據(jù)塊。Federation架構(gòu)3、每個block pool內(nèi)部自治，也就是說各自管理各自的block，不會與其他block pool交流。一個namenod