HDFS2.X中NameNode塊報(bào)告處理的示例分析-創(chuàng)新互聯(lián)

這篇文章主要介紹了HDFS2.X中NameNode塊報(bào)告處理的示例分析，具有一定借鑒價(jià)值，感興趣的朋友可以參考下，希望大家閱讀完這篇文章之后大有收獲，下面讓小編帶著大家一起了解一下。

成都創(chuàng)新互聯(lián)：從2013年開(kāi)始為各行業(yè)開(kāi)拓出企業(yè)自己的“網(wǎng)站建設(shè)”服務(wù)，為數(shù)千家公司企業(yè)提供了專業(yè)的成都做網(wǎng)站、成都網(wǎng)站設(shè)計(jì)、網(wǎng)頁(yè)設(shè)計(jì)和網(wǎng)站推廣服務(wù)， 按需網(wǎng)站建設(shè)由設(shè)計(jì)師親自精心設(shè)計(jì)，設(shè)計(jì)的效果完全按照客戶的要求，并適當(dāng)?shù)奶岢龊侠淼慕ㄗh，擁有的視覺(jué)效果，策劃師分析客戶的同行競(jìng)爭(zhēng)對(duì)手，根據(jù)客戶的實(shí)際情況給出合理的網(wǎng)站構(gòu)架，制作客戶同行業(yè)具有領(lǐng)先地位的。

NameNode會(huì)接收兩種情況的塊報(bào)告，DataNode全部塊報(bào)告與增量塊報(bào)告。

4.1全量報(bào)告分析

       目前全量報(bào)告以周期性進(jìn)行報(bào)告，既然已經(jīng)有啟動(dòng)時(shí)候的全量數(shù)據(jù)塊報(bào)告，錯(cuò)誤塊報(bào)告，增量塊報(bào)告（包括刪除塊報(bào)告），為什么還需要周期性全量塊報(bào)告呢?比如某DataNode接受到數(shù)據(jù)塊但是增量報(bào)告失敗，那就需要周期性報(bào)告來(lái)解決了，或者NameNode給DN發(fā)送了刪除塊的命令，但是由于網(wǎng)絡(luò)等異常，DN沒(méi)收收到刪除命令，這樣DN再把這些數(shù)據(jù)塊報(bào)告上來(lái)就是無(wú)效塊，需要再次放入無(wú)效隊(duì)列，下次心跳再命令DN刪除；同時(shí)比如每次塊報(bào)告會(huì)清理DatanodeDescriptor對(duì)象維護(hù)的塊列表還有某個(gè)塊的信息，但是DN節(jié)點(diǎn)再也沒(méi)有報(bào)告上來(lái)，定時(shí)清除這些無(wú)效信息，有助于提高塊列表的操作性能，從而提供NameNode的性能。同時(shí)我們可以考慮分析是否還有其他原因可能影響NameNode的性能。

為了提高HDFS啟動(dòng)速度，在Hadoop2.0版本中全量塊報(bào)告分為了兩種：?jiǎn)?dòng)時(shí)候塊報(bào)告與非啟動(dòng)的時(shí)候塊報(bào)告，即是否是第一次塊報(bào)告。那么具體又是如何來(lái)提高啟動(dòng)速度的呢？在啟動(dòng)的時(shí)候，不計(jì)算哪些文件元數(shù)據(jù)需要?jiǎng)h除，不計(jì)算無(wú)效快，這些處理都推遲到下一次塊報(bào)告進(jìn)行處理

對(duì)于第一次塊報(bào)告，代碼調(diào)用流程為：NameNodeRpcServer.blockReport()->BlockManager. processReport()->BlockManager.processFirstBlockReport().對(duì)Standby節(jié)點(diǎn)，如果報(bào)告的數(shù)據(jù)塊所相關(guān)元數(shù)據(jù)日志從節(jié)點(diǎn)還沒(méi)有加載完畢，則會(huì)將報(bào)告的塊信息加入一個(gè)隊(duì)列，當(dāng)Standby節(jié)點(diǎn)加載元數(shù)據(jù)后，再處理該消息隊(duì)列，第一次塊報(bào)告處理詳細(xì)代碼如下，可以看到，為了提高報(bào)告速度，只有簡(jiǎn)單的幾步進(jìn)行塊報(bào)告處理，僅有驗(yàn)證塊是否損壞，然后直接判斷塊狀態(tài)是否為FINALIZED狀態(tài)，如果是，就直接建立塊與DN節(jié)點(diǎn)的映射。

[java] view plain copy

  1. private void processFirstBlockReport(final DatanodeDescriptor node,  

  2.       final BlockListAsLongs report) throws IOException {  

  3.     if (report == null) return;  

  4.     assert (namesystem.hasWriteLock());  

  5.     assert (node.numBlocks() == 0);  

  6.     BlockReportIterator itBR = report.getBlockReportIterator();  

  7.    

  8.     while(itBR.hasNext()) {  

  9.       Block iblk = itBR.next();  

  10.       ReplicaState reportedState = itBR.getCurrentReplicaState();  

  11.       //對(duì)于從節(jié)點(diǎn)shouldPostponeBlocksFromFuture為true；判斷塊時(shí)間戳//是否大于目前時(shí)間  

  12.       if (shouldPostponeBlocksFromFuture&&  

  13.           namesystem.isGenStampInFuture(iblk.getGenerationStamp())) {  

  14. //將塊信息加入隊(duì)列，從節(jié)點(diǎn)消化完相關(guān)日志，會(huì)處理該隊(duì)列  

  15.         queueReportedBlock(node, iblk, reportedState,  

  16.             QUEUE_REASON_FUTURE_GENSTAMP);  

  17.         continue;  

  18.       }  

  19.        

  20.       BlockInfo storedBlock = blocksMap.getStoredBlock(iblk);  

  21.       // If block does not belong to any file, we are done.  

  22.       if (storedBlock == null) continue;  

  23.        

  24.       // If block is corrupt, mark it and continue to next block.  

  25.       BlockUCState ucState = storedBlock.getBlockUCState();  

  26.       BlockToMarkCorrupt c = checkReplicaCorrupt(  

  27.           iblk, reportedState, storedBlock, ucState, node);  

  28.       if (c != null) {  

  29. //對(duì)于從節(jié)點(diǎn)，先將塊信息加入pendingDNMessages隊(duì)列  

  30. //將塊信息加入隊(duì)列，從節(jié)點(diǎn)消化完相關(guān)日志，會(huì)處理該隊(duì)列，如果該塊還是被損壞，就真的是損壞了  

  31.         if (shouldPostponeBlocksFromFuture) {  

  32.           // In the Standby, we may receive a block report for a file that we  

  33.           // just have an out-of-date gen-stamp or state for, for example.  

  34.           queueReportedBlock(node, iblk, reportedState,  

  35.               QUEUE_REASON_CORRUPT_STATE);  

  36.         } else {  

  37. //對(duì)于主節(jié)點(diǎn)，有塊損壞，直接標(biāo)記為損壞  

  38.           markBlockAsCorrupt(c, node);  

  39.         }  

  40.         continue;  

  41.       }  

  42.        

  43.       // If block is under construction, add this replica to its list  

  44.       if (isBlockUnderConstruction(storedBlock, ucState, reportedState)) {  

  45.         ((BlockInfoUnderConstruction)storedBlock).addReplicaIfNotPresent(  

  46.             node, iblk, reportedState);  

  47.         //and fall through to next clause  

  48.       }       

  49.       //add replica if appropriate  

  50.       if (reportedState == ReplicaState.FINALIZED) {  

  51.         addStoredBlockImmediate(storedBlock, node);  

  52.       }  

  53.     }  

  54.   }  

而對(duì)于非第一次塊報(bào)告，情況就要復(fù)雜一些了，對(duì)于報(bào)告的每個(gè)塊信息，不僅會(huì)建立塊與DN的映射，而且均會(huì)檢查塊是否損壞，塊是是否無(wú)效，元數(shù)據(jù)是否已經(jīng)無(wú)效應(yīng)該刪除，是否為UC狀態(tài)的塊等，該過(guò)程主要由方法processReport來(lái)完成

[java] view plain copy

  1. private void processReport(final DatanodeDescriptor node,  

  2.       final BlockListAsLongs report) throws IOException {  

  3.     // Normal case:  

  4.     // Modify the (block-->datanode) map, according to the difference  

  5.     // between the old and new block report.  

  6.     //  

  7.     Collection<BlockInfo> toAdd = new LinkedList<BlockInfo>();  

  8.     Collection<Block> toRemove = new LinkedList<Block>();  

  9.     Collection<Block> toInvalidate = new LinkedList<Block>();  

  10.     Collection<BlockToMarkCorrupt> toCorrupt = new LinkedList<BlockToMarkCorrupt>();  

  11. Collection<StatefulBlockInfo> toUC = new LinkedList<StatefulBlockInfo>();  

  12. //統(tǒng)計(jì)塊，并且判斷塊是否應(yīng)該刪除，是否應(yīng)該添加到blocksMap列表等  

  13.     reportDiff(node, report, toAdd, toRemove, toInvalidate, toCorrupt, toUC);  

  14.    

  15.     // Process the blocks on each queue  

  16.     for (StatefulBlockInfo b : toUC) {  

  17.       addStoredBlockUnderConstruction(b.storedBlock, node, b.reportedState);  

  18.     }  

  19.     for (Block b : toRemove) {  

  20.       removeStoredBlock(b, node);  

  21.     }  

  22.     for (BlockInfo b : toAdd) {  

  23.       addStoredBlock(b, node, null, true);  

  24.     }  

  25.     for (Block b : toInvalidate) {  

  26.       NameNode.stateChangeLog.info("BLOCK* processReport: block "  

  27.           + b + " on " + node + " size " + b.getNumBytes()  

  28.           + " does not belong to any file.");  

  29.       addToInvalidates(b, node);  

  30.     }  

  31.     for (BlockToMarkCorrupt b : toCorrupt) {  

  32.       markBlockAsCorrupt(b, node);  

  33.     }  

  34.   }  

在reportDiff方法內(nèi)，實(shí)現(xiàn)如下：

[java] view plain copy

  1. private void reportDiff(DatanodeDescriptor dn,  

  2.       BlockListAsLongs newReport,  

  3.       Collection<BlockInfo> toAdd,              // add to DatanodeDescriptor  

  4.       Collection<Block> toRemove,           // remove from DatanodeDescriptor  

  5.       Collection<Block> toInvalidate,       // should be removed from DN  

  6.       Collection<BlockToMarkCorrupt> toCorrupt, // add to corrupt replicas list  

  7.       Collection<StatefulBlockInfo> toUC) { // add to under-construction list  

  8.     // place a delimiter分隔符 in the list which separates blocks  

  9.     // that have been reported from those that have not  

  10.     BlockInfo delimiter = new BlockInfo(new Block(), 1);  

  11.     boolean added = dn.addBlock(delimiter);  

  12.     assert added : "Delimiting block cannot be present in the node";  

  13.     int headIndex = 0; //currently the delimiter is in the head of the list  

  14.     int curIndex;  

  15.    

  16.     if (newReport == null)  

  17.       newReport = new BlockListAsLongs();  

  18.     // scan the report and process newly reported blocks  

  19.     BlockReportIterator itBR = newReport.getBlockReportIterator();  

  20.     while(itBR.hasNext()) {  

  21.       Block iblk = itBR.next();  

  22.       ReplicaState iState = itBR.getCurrentReplicaState();  

  23.       BlockInfo storedBlock = processReportedBlock(dn, iblk, iState,  

  24.                                   toAdd, toInvalidate, toCorrupt, toUC);  

  25.       // move block to the head of the list  

  26.       if (storedBlock != null && (curIndex = storedBlock.findDatanode(dn)) >= 0) {  

  27.         headIndex = dn.moveBlockToHead(storedBlock, curIndex, headIndex);  

  28.       }  

  29.     }  

  30.     // collect blocks that have not been reported  

  31.     // all of them are next to the delimiter  

  32.     //收集DN對(duì)象中所有沒(méi)有被DN節(jié)點(diǎn)報(bào)告上來(lái)的塊，將這些塊信息從DN對(duì)象維護(hù)的列表中刪除，這樣可以有效控制DN塊列表中存在大量的無(wú)效塊，  

  33.     //影響NameNode的操作性能  

  34.     Iterator<? extends Block> it = new DatanodeDescriptor.BlockIterator(  

  35.         delimiter.getNext(0), dn);  

  36.     while(it.hasNext())  

  37.       toRemove.add(it.next());  

  38.     dn.removeBlock(delimiter);  

  39.   }  

4.2增量報(bào)告分析

     相比于全量塊報(bào)告方式，增量報(bào)告報(bào)告DN節(jié)點(diǎn)很短時(shí)間內(nèi)已經(jīng)接收完成，或者正在接受或者刪除的塊，而且為了提高文件上傳的效率， DN節(jié)點(diǎn)應(yīng)該盡快將接受到的塊報(bào)告給NameNode，現(xiàn)在引入了RECEIVING_BLOCK這個(gè)一個(gè)塊狀態(tài)，有可能就是為了提高寫(xiě)入速度。

HDFS2.X中NameNode塊報(bào)告處理的示例分析

                                                                     增量塊報(bào)告流程圖

正在接收的塊與已經(jīng)接收完的塊，除了在數(shù)據(jù)塊狀態(tài)不一樣外，其他基本相同，其接收塊代碼調(diào)用流程如下：NameNodeRpcServer.blockReceivedAndDeleted()->BlockManager.processIncrementalBlockReport()->BlockManager. addBlock()->BlockManager.processAndHandleReportedBlock()->BlockManager.processReportedBlock(),在方法processReportedBlock中，首先判斷報(bào)告的塊是否元數(shù)據(jù)已經(jīng)從主節(jié)點(diǎn)讀取到，如果沒(méi)有加入消息列表

[java] view plain copy

  1. //postpone延期  

  2.     //如果是從節(jié)點(diǎn)，可能雖然DN節(jié)點(diǎn)將塊信息報(bào)告上來(lái)，但是元數(shù)據(jù)還沒(méi)有從日志中消化到  

  3.     if (shouldPostponeBlocksFromFuture &&  

  4.         namesystem.isGenStampInFuture(block.getGenerationStamp())) {  

  5.       queueReportedBlock(dn, block, reportedState,  

  6.           QUEUE_REASON_FUTURE_GENSTAMP);  

  7.       return null;  

  8.     }  

<br font-size:16px;white-space:normal;background-color:#FFFFFF;" />然后從blocksMap中查詢到數(shù)據(jù)塊對(duì)于文件inode，判斷文件是否存在；如果判斷塊屬于損害塊，冗余分?jǐn)?shù)是否不夠等情況，如果塊一切正常，且狀態(tài)為完成，將將其加入blocksMap等集合列表。具體代碼如下：

[java] view plain copy

  1. //檢查塊是否已經(jīng)被損害  

  2.     BlockToMarkCorrupt c = checkReplicaCorrupt(  

  3.         block, reportedState, storedBlock, ucState, dn);  

  4.     if (c != null) {  

  5.       if (shouldPostponeBlocksFromFuture) {  

  6.         // If the block is an out-of-date generation stamp or state,  

  7.         // but we're the standby, we shouldn't treat it as corrupt,  

  8.         // but instead just queue it for later processing.  

  9.         queueReportedBlock(dn, storedBlock, reportedState,  

  10.             QUEUE_REASON_CORRUPT_STATE);  

  11.       } else {  

  12.           //將其加入損害列表  

  13.         toCorrupt.add(c);  

  14.       }  

  15.       return storedBlock;  

  16.     }  

  17.      //如果該數(shù)據(jù)塊正在被構(gòu)建，加入構(gòu)建列表  

  18.     if (isBlockUnderConstruction(storedBlock, ucState, reportedState)) {  

  19.       toUC.add(new StatefulBlockInfo(  

  20.           (BlockInfoUnderConstruction)storedBlock, reportedState));  

  21.       return storedBlock;  

  22.     }  

  23.    

  24.     //add replica if appropriate  

  25.     //如果報(bào)告的塊狀態(tài)為FINALIZED且該DN沒(méi)有報(bào)告該塊，則加入添加隊(duì)列  

  26.     if (reportedState == ReplicaState.FINALIZED  

  27.         && storedBlock.findDatanode(dn) < 0) {  

  28.       toAdd.add(storedBlock);  

  29.     }  

  30.     return storedBlock;  

感謝你能夠認(rèn)真閱讀完這篇文章，希望小編分享的“HDFS2.X中NameNode塊報(bào)告處理的示例分析”這篇文章對(duì)大家有幫助，同時(shí)也希望大家多多支持創(chuàng)新互聯(lián)，關(guān)注創(chuàng)新互聯(lián)-成都網(wǎng)站建設(shè)公司行業(yè)資訊頻道，更多相關(guān)知識(shí)等著你來(lái)學(xué)習(xí)!

本文題目：HDFS2.X中NameNode塊報(bào)告處理的示例分析-創(chuàng)新互聯(lián)
文章路徑：http://muchs.cn/article2/djjsic.html

成都網(wǎng)站建設(shè)公司_創(chuàng)新互聯(lián)，為您提供小程序開(kāi)發(fā)、定制開(kāi)發(fā)、手機(jī)網(wǎng)站建設(shè)、外貿(mào)建站、響應(yīng)式網(wǎng)站、云服務(wù)器

廣告

聲明：本網(wǎng)站發(fā)布的內(nèi)容（圖片、視頻和文字）以用戶投稿、用戶轉(zhuǎn)載內(nèi)容為主，如果涉及侵權(quán)請(qǐng)盡快告知，我們將會(huì)在第一時(shí)間刪除。文章觀點(diǎn)不代表本網(wǎng)站立場(chǎng)，如需處理請(qǐng)聯(lián)系客服。電話：028-86922220；郵箱：631063699@qq.com。內(nèi)容未經(jīng)允許不得轉(zhuǎn)載，或轉(zhuǎn)載時(shí)需注明來(lái)源： 創(chuàng)新互聯(lián)

成都網(wǎng)站建設(shè)