帶你入坑大數(shù)據(jù)（二） --- HDFS的讀寫流程和一些重要策略

## 前言 ### 前情回顧 如果說(shuō)上一篇是在闡述HDFS最基礎(chǔ)的理論知識(shí)，這一篇就是HDFS的主要工作流程，和一些較為有用的策略 補(bǔ)充一個(gè)問(wèn)題，就是當(dāng)我們 NameNode 掛掉，SecondaryNameNode作為新的NameNode上位時(shí)，它確實(shí)可以根據(jù)fsimage.ckpt把一部分元數(shù)據(jù)加載到內(nèi)存，可是如果這時(shí)還有一部分操作日志在edits new中沒(méi)有執(zhí)行怎么辦？ 這時(shí)候有一個(gè)解決方案就是利用一個(gè)network fileSystem來(lái)解決，比如說(shuō)集群中有一個(gè)服務(wù)器安裝了一個(gè)nfs server，而在NameNode上再安裝一個(gè)nfs client，此時(shí)客戶端向HDFS寫數(shù)據(jù)時(shí)，同時(shí)把向edits new中寫的數(shù)據(jù)寫一份到nfs server中，SecondaryNamenode就可以通過(guò)這個(gè)nfs server來(lái)獲取此時(shí)斷層的數(shù)據(jù)了 其他似乎也沒(méi)啥可多說(shuō)的，讓我們直奔主題吧

 以往鏈接 [從零開(kāi)始的大數(shù)據(jù)（一） --- HDFS的知識(shí)概述（上）] ## 一、HDFS的讀流程 之后的內(nèi)容會(huì)圍繞下圖開(kāi)始  

 1.認(rèn)識(shí)角色 簡(jiǎn)單過(guò)一下圖里面的角色，最大塊的是一個(gè)client node，也就是說(shuō)，這個(gè)節(jié)點(diǎn)上運(yùn)行著客戶端，如果實(shí)在是沒(méi)搞清楚哪個(gè)是客戶端，那也很簡(jiǎn)單，平時(shí)沒(méi)事就執(zhí)行 hadoop fs -ls / 這個(gè)命令的機(jī)器，那就是客戶端了，其他就是NameNode和DataNode，在client node上運(yùn)行著一個(gè)JVM虛擬機(jī)，讓HDFS client跑起來(lái) 

 2.步驟分析 

 ① HDFS client調(diào)用文件系統(tǒng)的open方法 Distributed FileSystem顧名思義是一個(gè)分布式文件系統(tǒng)，它會(huì)通過(guò)RPC的方式遠(yuǎn)程過(guò)程調(diào)用**NameNode里的open方法**，這個(gè)open方法有什么作用呢，就是獲取要讀的文件的**file block locations**，也就是文件的block的位置，在上一講我們也已經(jīng)提到了，一個(gè)文件是會(huì)分割成128M一塊的大小分別存儲(chǔ)在各個(gè)數(shù)據(jù)節(jié)點(diǎn)的。 同時(shí)在執(zhí)行open方法時(shí)，客戶端會(huì)產(chǎn)生一個(gè)FSData InputStream的一個(gè)輸入流對(duì)象（客戶端讀數(shù)據(jù)是從外部讀回來(lái)的） 

 ② FSData InputStream讀數(shù)據(jù) HDFS client調(diào)用FSData InputStream的read方法，同上也是遠(yuǎn)程過(guò)程**調(diào)用DataNode的read方法**，此時(shí)的讀取順序是由近到遠(yuǎn)，就是DataNode和client node的距離，這里所指的距離是一種物理距離，判定可以參考上一篇文章中機(jī)架的概念。 在聯(lián)系上DataNode并成功讀取后，關(guān)閉流就走完了一個(gè)正常的流程。 而且補(bǔ)充一下就是，上面Distributed FileSystem所調(diào)用的get block locations的方法只會(huì)返回部分?jǐn)?shù)據(jù)塊，get block locations會(huì)分批次地返回block塊的位置信息。讀block塊理論上來(lái)說(shuō)是依次讀，當(dāng)然也可以通過(guò)多線程的方式實(shí)現(xiàn)同步讀。 

 ③ 容錯(cuò)機(jī)制 

 1.如果client從DataNode上讀取block時(shí)網(wǎng)絡(luò)中斷了如何解決？ 此時(shí)我們會(huì)找到block另外的副本（一個(gè)block塊有3個(gè)副本，上一篇已經(jīng)說(shuō)過(guò)了），并且通過(guò)FSData InputStream進(jìn)行記錄，以后就不再?gòu)闹袛嗟母北旧献x了。 

 2.如果一個(gè)DataNode掛掉了怎么辦？ 在上一篇中我們提到了一個(gè)HDFS的心跳機(jī)制，DataNode會(huì)隔一小時(shí)向NameNode匯報(bào)blockReport，比如現(xiàn)在的情況是，block1的三個(gè)副本分別存儲(chǔ)在DataNode1,2,3上，此時(shí)DataNode1掛掉了。NameNode得知某個(gè)block還剩2個(gè)副本，此時(shí)攜帶這block的其余兩個(gè)副本的DataNode2,3在向NameNode報(bào)告時(shí)，NameNode就會(huì)對(duì)它們中的某一個(gè)返回一個(gè)指令，把block1復(fù)制一份給其他正常的節(jié)點(diǎn)。讓block1恢復(fù)成原本的3個(gè)副本。 

 3.client如何保證讀取數(shù)據(jù)的完整性 因?yàn)閺腄ataNode上讀數(shù)據(jù)是通過(guò)網(wǎng)絡(luò)來(lái)讀取的，這說(shuō)明會(huì)存在讀取過(guò)來(lái)的數(shù)據(jù)是不完整的或者是錯(cuò)誤的情況。 DataNode上存儲(chǔ)的不僅僅是數(shù)據(jù)，數(shù)據(jù)還附帶著一個(gè)叫做checkSum檢驗(yàn)和（CRC32算法）的概念，針對(duì)于任何大小的數(shù)據(jù)塊計(jì)算CRC32的值都是32位4個(gè)字節(jié)大小。此時(shí)我們的FSData InputStream向DataNode讀數(shù)據(jù)時(shí)，會(huì)將與這份數(shù)據(jù)對(duì)應(yīng)的checkSum也一并讀取過(guò)來(lái)，此時(shí)FSData InputStream再對(duì)它讀過(guò)來(lái)的數(shù)據(jù)做一個(gè)checkSum，把它與讀過(guò)來(lái)的checkSum做一個(gè)對(duì)比，如果不一致，就重新從另外的DataNode上再次讀取。 

4.上一個(gè)問(wèn)題完成后工作 FSData InputStream會(huì)告訴NameNode，這個(gè)DataNode上的這個(gè)block有問(wèn)題了，NameNode收到消息后就會(huì)再通過(guò)心跳機(jī)制通知這個(gè)DataNode刪除它的block塊，然后再用類似2的做法，讓正常的DataNode去copy一份正常的block數(shù)據(jù)給其它節(jié)點(diǎn)，保證副本數(shù)為3 代碼簡(jiǎn)單示例（可跳過(guò)） try { // String srcFile = "hdfs://node-01:9000/data/hdfs01.mp4"; Configuration conf = new Configuration(); FileSystem fs = FileSystem.get(URI.create(srcFile),conf); FSDataInputStream hdfsInStream = fs.open(new Path(srcFile)); BufferedOutputStream outputStream = new BufferedOutputStream(new FileOutputStream("/home/node1/hdfs02.mp4")); IOUtils.copyBytes(hdfsInStream, outputStream, 4096, true); } catch (IOException e) { e.printStackTrace(); } ## 二、HDFS寫流程 寫流程我們會(huì)按照下圖來(lái)進(jìn)行講解，比讀數(shù)據(jù)更加復(fù)雜一丟丟，角色基本沒(méi)有改變所以就不詳細(xì)介紹了  客戶端向HDFS寫數(shù)據(jù)的時(shí)候是把文件分塊存儲(chǔ)在HDFS的各個(gè)節(jié)點(diǎn)上，而規(guī)定了存儲(chǔ)位置的是NameNode，所以Client node在存儲(chǔ)文件時(shí)需要先和NameNode進(jìn)行聯(lián)系讓它進(jìn)行分配。 

步驟分析

 ① 客戶端調(diào)用分布式文件系統(tǒng)的create方法 和上面讀的方法類似，不過(guò)這次調(diào)用的是Distributed FileSystem的create方法，此時(shí)也是通過(guò)遠(yuǎn)程調(diào)用NameNode的create方法 此時(shí)NameNode會(huì)進(jìn)行的舉措 *

 1.檢測(cè)自己是否正常運(yùn)行

 2.判斷要?jiǎng)?chuàng)建的文件是否存在

 3.

client是否有創(chuàng)建文件的權(quán)限

 4.

對(duì)HDFS做狀態(tài)的更改需要在edits log寫日志記錄

 ② 客戶端調(diào)用輸出流的write方法 create方法的返回值是一個(gè)OutputStream對(duì)象，為什么是output，因?yàn)槭怯蒆DFS去往DataNode去寫數(shù)據(jù)，此時(shí)HDFS會(huì)調(diào)用這個(gè)OutputStream的write方法 但是有個(gè)問(wèn)題，此時(shí)我們還不知道我們的這些block塊要分別存放于哪些節(jié)點(diǎn)上，所以此時(shí)FSData OutputStream就要再和NameNode交互一下，遠(yuǎn)程過(guò)程調(diào)用**NameNode的addBlock**方法，這個(gè)方法**返回的是各個(gè)block塊分別需要寫在哪3個(gè)DataNode**上面。 此時(shí)OutputStream就完整得知了數(shù)據(jù)和數(shù)據(jù)該往哪里去寫了 

 ③ 具體的寫流程分析 請(qǐng)看流程4.1，**chunk**是一個(gè)512字節(jié)大小的數(shù)據(jù)塊，寫數(shù)據(jù)的過(guò)程中數(shù)據(jù)是一字節(jié)一字節(jié)往chunk那里寫的，當(dāng)寫滿一個(gè)chunk后，會(huì)計(jì)算一個(gè)checkSum，這個(gè)checkSum是4個(gè)字節(jié)大小，計(jì)算完成后一并放入chunk，所以整一個(gè)**chunk大小其實(shí)是512字節(jié)+4字節(jié)=516字節(jié)**。 上述步驟結(jié)束后，一個(gè)chunk就會(huì)往**package**里面放，package是一個(gè)**64kb大小的數(shù)據(jù)包**，我們知道 64kb = 64 * 1024字節(jié)，所以這個(gè)package可以放非常多的chunk。 此時(shí)一個(gè)package滿了之后，會(huì)把這個(gè)packjage放到一個(gè)data queue隊(duì)列里面，之后會(huì)陸續(xù)有源源不斷的package傳輸過(guò)來(lái)，圖中用p1，p2···等表示 這時(shí)候開(kāi)始真正的寫數(shù)據(jù)過(guò)程 

1. data queue中的package往數(shù)據(jù)節(jié)點(diǎn)DataNode上傳輸，傳輸?shù)捻樞虬凑誑ameNode的addBlock()方法返回的列表依次傳輸** （ps：傳輸?shù)念悶橐粋€(gè)叫做dataStreamer的類，而且其實(shí)addBlock方法返回的列表基本是按照離客戶端物理距離由近到遠(yuǎn)的順序的）

 2. 往DataNode上傳輸?shù)耐瑫r(shí)也往確認(rèn)隊(duì)列ack queue上傳輸

3. 針對(duì)DataNode中傳輸完成的數(shù)據(jù)做一個(gè)checkSum，并與原本打包前的checkSum做一個(gè)比較

 4. 校驗(yàn)成功，就從確認(rèn)隊(duì)列ack queue中刪除該package，否則該package重新置入data queue重傳

 補(bǔ)充：

1.以上邏輯歸屬于FSData OutputStream的邏輯 

2.雖然本身一個(gè)block為128M，而package為64Kb，128M對(duì)于網(wǎng)絡(luò)傳輸過(guò)程來(lái)說(shuō)算是比較大，拆分為小包是為了可靠傳輸 

3.網(wǎng)絡(luò)中斷時(shí)的舉措：HDFS會(huì)先把整個(gè)pineline關(guān)閉，然后獲取一個(gè)已存在的完整的文件的version，發(fā)送給NameNode后，由NameNode通過(guò)心跳機(jī)制對(duì)未正確傳輸?shù)臄?shù)據(jù)下達(dá)刪除命令

 4.如果是某個(gè)DataNode不可用，在1中我們也提到過(guò)了，通過(guò)心跳機(jī)制會(huì)通知其余的可用DataNode的其中一個(gè)進(jìn)行copy到一個(gè)可用節(jié)點(diǎn)上 

 ④ 寫入結(jié)束后的行動(dòng) 完成后通過(guò)心跳機(jī)制NameNode就可以得知副本已經(jīng)創(chuàng)建完成，再調(diào)用addBlock()方法寫之后的文件。 

⑤ 流程總結(jié) **1.client端調(diào)用Distributed FileSystem的create，此時(shí)是遠(yuǎn)程調(diào)用了NameNode的create，此時(shí)NameNode進(jìn)行4個(gè)操作，檢測(cè)自己是否正常，文件是否存在，客戶端的權(quán)限和寫日志

2.create的返回值為一個(gè)FSData OutputStream對(duì)象，此時(shí)client調(diào)用流的write方法，和NameNode進(jìn)行連接，NameNode的addBlock方法返回塊分配的DataNode列表

3.開(kāi)始寫數(shù)據(jù)，先寫在chunk，后package，置入data queue，此時(shí)兩個(gè)操作，向DataNode傳輸，和放入ack queue，DataNode傳輸結(jié)束會(huì)檢測(cè)checkSum，成功就刪除ack queue的package，否則放回data queue重傳

4.結(jié)束后關(guān)閉流，告訴NameNode，調(diào)用complete方法結(jié)束** #### 簡(jiǎn)單代碼示例（可跳過(guò)） String source="/home/node1/hdfs01.mp4"; //linux中的文件路徑,demo存在一定數(shù)據(jù) //先確保/data目錄存在 String destination="hdfs://node-01:9000/data/hdfs01.mp4";//HDFS的路徑 InputStream in = null; try { in = new BufferedInputStream(new FileInputStream(source)); //HDFS讀寫的配置文件 Configuration conf = new Configuration(); FileSystem fs = FileSystem.get(URI.create(destination),conf); //調(diào)用Filesystem的create方法返回的是FSDataOutputStream對(duì)象 //該對(duì)象不允許在文件中定位，因?yàn)镠DFS只允許一個(gè)已打開(kāi)的文件順序?qū)懭牖蜃芳? OutputStream out = fs.create(new Path(destination)); IOUtils.copyBytes(in, out, 4096, true); } catch (FileNotFoundException e) { System.out.println("exception"); e.printStackTrace(); } catch (IOException e) { System.out.println("exception1"); e.printStackTrace(); } ## 3.Hadoop HA高可用  之前已經(jīng)提到過(guò)，元數(shù)據(jù)是放在NameNode的內(nèi)存中的，當(dāng)元數(shù)據(jù)丟失，造成服務(wù)不可用，這時(shí)候就需要時(shí)間來(lái)恢復(fù)。HA就可以讓用戶感知不到這種問(wèn)題。

這需要yarn，MapReduce，zookeeper的支持 僅有一個(gè)NameNode時(shí)，當(dāng)NameNode掛掉就需要把fsimage讀到內(nèi)存，然后把edits log所記錄的日志重新執(zhí)行一遍，元數(shù)據(jù)才能恢復(fù)，而這種做法需要大量的時(shí)間 所以解決方案就在于我們要花大量時(shí)間來(lái)恢復(fù)元數(shù)據(jù)metaData，所以解決的方案就是讓集群瞬間變回可用狀態(tài)即可。 

通過(guò)設(shè)置一個(gè)stand by的NameNode，并和主NameNode的元數(shù)據(jù)保持一致，圖中綠色的區(qū)域表示一個(gè)共享存儲(chǔ)，主NameNode的元數(shù)據(jù)會(huì)傳輸至共享存儲(chǔ)里面，讓stand by的NameNode進(jìn)行同步。 下面的DataNode會(huì)同時(shí)往兩個(gè)NameNode發(fā)送blockReport，因?yàn)樽x取DataNode的塊信息并不會(huì)很快，所以為了保證在active掛掉的時(shí)候，standby能立刻頂上位置，所以要事先讀取塊信息，同時(shí)這也是方便standby來(lái)構(gòu)建它的元數(shù)據(jù)信息的途徑。 

active掛掉后讓stand by立刻生效的機(jī)制是上面的FailoverControllerActive實(shí)現(xiàn)的，簡(jiǎn)稱zkfc，它會(huì)定時(shí)ping主NameNode，如果發(fā)現(xiàn)NameNode掛掉，就會(huì)通知我們的zookeeper集群，然后集群的另一個(gè)FailoverControllerActive就會(huì)通知stand by。 ## 4.Hadoop聯(lián)邦 集群中的元數(shù)據(jù)會(huì)保存在NameNode的內(nèi)存中，而這些元數(shù)據(jù)每份占用約150字節(jié)，對(duì)于一個(gè)擁有大量文件的集群來(lái)說(shuō)，因?yàn)镹ameNode的metaData被占滿，DataNode就無(wú)法寫入了，聯(lián)邦就可以幫助系統(tǒng)突破文件數(shù)上限 其實(shí)就是布置了多個(gè)NameNode來(lái)共同維護(hù)集群，來(lái)增加namespace，而且分散了NameNode的訪問(wèn)壓力，而且客戶端的讀寫互不影響。就是**擴(kuò)展性，高吞吐和隔離性**。

  ## 5.HDFS存儲(chǔ)大量小文件 和剛剛的聯(lián)邦的介紹時(shí)的情況一樣，文件數(shù)量（每個(gè)文件元數(shù)據(jù)150byte）會(huì)影響到NameNode的內(nèi)存 ### 方案1：HAR文件方案 其實(shí)就是通過(guò)一個(gè)MR程序把許多小文件合并成一個(gè)大文件，需要啟動(dòng)Yarn # 創(chuàng)建archive文件 hadoop archive -archiveName test.har -p /testhar -r 3 th1 th2 /outhar # 原文件還存在，需手動(dòng)刪除 # 查看archive文件 hdfs dfs -ls -R har:///outhar/test.har # 解壓archive文件 hdfs dfs -cp har:///outhar/test.har/th1 hdfs:/unarchivef hadoop fs -ls /unarchivef # 順序 hadoop distcp har:///outhar/test.har/th1 hdfs:/unarchivef2 # 并行，啟動(dòng)MR ### 方案2：Sequence File方案 其核心是以文件名為key，文件內(nèi)容為value組織小文件。

10000個(gè)100KB的小文件，可以編寫程序?qū)⑦@些文件放到一個(gè)SequenceFile文件，然后就以數(shù)據(jù)流的方式處理這些文件，也可以使用MapReduce進(jìn)行處理。一個(gè)SequenceFile是可分割的，所以MapReduce可將文件切分成塊，每一塊獨(dú)立操作。不像HAR,SequenceFile支持壓縮。在大多數(shù)情況下，以block為單位進(jìn)行壓縮是最好的選擇，因?yàn)橐粋€(gè)block包含多條記錄，壓縮作用在block之上，比reduce壓縮方式（一條一條記錄進(jìn)行壓縮）的壓縮比高.把已有的數(shù)據(jù)轉(zhuǎn)存為SequenceFile比較慢。比起先寫小文件，再將小文件寫入SequenceFile，一個(gè)更好的選擇是直接將數(shù)據(jù)寫入一個(gè)SequenceFile文件，省去小文件作為中間媒介. 此方案的代碼不是很重要，所以就直接省略了，實(shí)在是想看看長(zhǎng)啥樣的可以艾特我 ## finally 關(guān)于HDFS比較細(xì)節(jié)的東西在這篇有補(bǔ)充 到此HDFS的內(nèi)容就差不多了，希望對(duì)你會(huì)有所幫助。之后會(huì)繼續(xù)往下分享MapReduce，帶你走完整個(gè)大數(shù)據(jù)的流程，感興趣的朋友可以持續(xù)關(guān)注下，謝謝。