• <ins id="pjuwb"></ins>
    <blockquote id="pjuwb"><pre id="pjuwb"></pre></blockquote>

    <noscript id="pjuwb"></noscript>
          <sup id="pjuwb"><pre id="pjuwb"></pre></sup>

            <dd id="pjuwb"></dd>
            <abbr id="pjuwb"></abbr>
            

    
            

            




            







												


        


            

            
	
	


            
	
            
		面對(duì)現(xiàn)實(shí)，超越自己
            
		逆水行舟，不進(jìn)則退
	
            

            
            		
	

            

            


            
	
		
            
			


            
posts - 269,comments - 32,trackbacks - 0


            
	

            


            
	
					


            
	
	1、NameNode啟動(dòng)加載元數(shù)據(jù)情景分析
            • NameNode函數(shù)里調(diào)用FSNamesystemm讀取dfs.namenode.name.dir和dfs.namenode.edits.dir構(gòu)建FSDirectory。
            • FSImage類(lèi)recoverTransitionRead和saveNameSpace分別實(shí)現(xiàn)了元數(shù)據(jù)的檢查、加載、內(nèi)存合并和元數(shù)據(jù)的持久化存儲(chǔ)。
            • saveNameSpace將元數(shù)據(jù)寫(xiě)入到磁盤(pán)，具體操作步驟：首先將current目錄重命名為lastcheckpoint.tmp;然后在創(chuàng)建新的current目錄，并保存文件；最后將lastcheckpoint.tmp重命名為privios.checkpoint.
            • checkPoint的過(guò)程：Secondary NameNode會(huì)通知nameNode產(chǎn)生一個(gè)edit log文件edits.new，之后所有的日志操作寫(xiě)入到edits.new文件中。接下來(lái)Secondary NameNode會(huì)從namenode下載fsimage和edits文件，進(jìn)行合并產(chǎn)生新的fsimage.ckpt;然后Secondary會(huì)將fsimage.ckpt文件上傳到namenode。最后namenode會(huì)重命名fsimage.ckpt為fsimage，edtis.new為edits；
            2、元數(shù)據(jù)更新及日志寫(xiě)入情景分析
            以mkdir為例:
            logSync代碼分析：
            代碼:
            1. public void logSync () throws IOException {  
            2. ArrayList<EditLogOutputStream > errorStreams = null ;  
            3. long syncStart = 0;  
            4.   
            5. // Fetch the transactionId of this thread.  
            6. long mytxid = myTransactionId .get (). txid;  
            7. EditLogOutputStream streams[] = null;  
            8. boolean sync = false;  
            9. try {  
            10. synchronized (this) {  
            11. assert editStreams. size() > 0 : "no editlog streams" ;  
            12. printStatistics (false);  
            13. // if somebody is already syncing, then wait  
            14. while (mytxid > synctxid && isSyncRunning) {  
            15. try {  
            16. wait (1000 );  
            17. } catch (InterruptedException ie ) {  
            18. }  
            19. }  
            20. //  
            21. // If this transaction was already flushed, then nothing to do  
            22. //  
            23. if (mytxid <= synctxid ) {  
            24. numTransactionsBatchedInSync ++;  
            25. if (metrics != null) // Metrics is non-null only when used inside name node  
            26. metrics .transactionsBatchedInSync .inc ();  
            27. return;  
            28. }  
            29. // now, this thread will do the sync  
            30. syncStart = txid ;  
            31. isSyncRunning = true;  
            32. sync = true;  
            33. // swap buffers  
            34. for( EditLogOutputStream eStream : editStreams ) {  
            35. eStream .setReadyToFlush ();  
            36. }  
            37. streams =  
            38. editStreams .toArray (new EditLogOutputStream[editStreams. size()]) ;  
            39. }  
            40. // do the sync  
            41. long start = FSNamesystem.now();  
            42. for (int idx = 0; idx < streams. length; idx++ ) {  
            43. EditLogOutputStream eStream = streams [idx ];  
            44. try {  
            45. eStream .flush ();  
            46. } catch (IOException ie ) {  
            47. FSNamesystem .LOG .error ("Unable to sync edit log." , ie );  
            48. //  
            49. // remember the streams that encountered an error.  
            50. //  
            51. if (errorStreams == null) {  
            52. errorStreams = new ArrayList <EditLogOutputStream >( 1) ;  
            53. }  
            54. errorStreams .add (eStream );  
            55. }  
            56. }  
            57. long elapsed = FSNamesystem.now() - start ;  
            58. processIOError (errorStreams , true);  
            59. if (metrics != null) // Metrics non-null only when used inside name node  
            60. metrics .syncs .inc (elapsed );  
            61. } finally {  
            62. synchronized (this) {  
            63. synctxid = syncStart ;  
            64. if (sync ) {  
            65. isSyncRunning = false;  
            66. }  
            67. this.notifyAll ();  
            68. }  
            69. }  
            70. }  

            3、Backup Node 的checkpoint的過(guò)程分析：
            1. /** 
            2. * Create a new checkpoint 
            3. */  
            4. void doCheckpoint() throws IOException {  
            5. long startTime = FSNamesystem.now ();  
            6. NamenodeCommand cmd =  
            7. getNamenode().startCheckpoint( backupNode. getRegistration());  
            8. CheckpointCommand cpCmd = null;  
            9. switch( cmd. getAction()) {  
            10. case NamenodeProtocol .ACT_SHUTDOWN :  
            11. shutdown() ;  
            12. throw new IOException ("Name-node " + backupNode .nnRpcAddress  
            13. + " requested shutdown.");  
            14. case NamenodeProtocol .ACT_CHECKPOINT :  
            15. cpCmd = (CheckpointCommand )cmd ;  
            16. break;  
            17. default:  
            18. throw new IOException ("Unsupported NamenodeCommand: "+cmd.getAction()) ;  
            19. }  
            20.   
            21. CheckpointSignature sig = cpCmd. getSignature();  
            22. assert FSConstants.LAYOUT_VERSION == sig .getLayoutVersion () :  
            23. "Signature should have current layout version. Expected: "  
            24. + FSConstants.LAYOUT_VERSION + " actual " + sig. getLayoutVersion();  
            25. assert !backupNode .isRole (NamenodeRole .CHECKPOINT ) ||  
            26. cpCmd. isImageObsolete() : "checkpoint node should always download image.";  
            27. backupNode. setCheckpointState(CheckpointStates .UPLOAD_START );  
            28. if( cpCmd. isImageObsolete()) {  
            29. // First reset storage on disk and memory state  
            30. backupNode. resetNamespace();  
            31. downloadCheckpoint(sig);  
            32. }  
            33.   
            34. BackupStorage bnImage = getFSImage() ;  
            35. bnImage. loadCheckpoint(sig);  
            36. sig.validateStorageInfo( bnImage) ;  
            37. bnImage. saveCheckpoint();  
            38.   
            39. if( cpCmd. needToReturnImage())  
            40. uploadCheckpoint(sig);  
            41.   
            42. getNamenode() .endCheckpoint (backupNode .getRegistration (), sig );  
            43.   
            44. bnImage. convergeJournalSpool();  
            45. backupNode. setRegistration(); // keep registration up to date  
            46. if( backupNode. isRole( NamenodeRole.CHECKPOINT ))  
            47. getFSImage() .getEditLog (). close() ;  
            48. LOG. info( "Checkpoint completed in "  
            49. + (FSNamesystem .now() - startTime )/ 1000 + " seconds."  
            50. + " New Image Size: " + bnImage .getFsImageName (). length()) ;  
            51. }  
            52. }  

            4、元數(shù)據(jù)可靠性機(jī)制。
            • 配置多個(gè)備份路徑。NameNode在更新日志或進(jìn)行Checkpoint的過(guò)程，會(huì)將元數(shù)據(jù)放在多個(gè)目錄下。
            • 對(duì)于沒(méi)一個(gè)需要保存的元數(shù)據(jù)文件，都創(chuàng)建一個(gè)輸出流，對(duì)訪(fǎng)問(wèn)過(guò)程中出現(xiàn)的異常輸出流進(jìn)行處理，將其移除。并再合適的時(shí)機(jī)再次檢查移除的數(shù)據(jù)量是否恢復(fù)正常。有效的保證了備份輸出流的異常問(wèn)題。
            • 采用了多種機(jī)制來(lái)保證元數(shù)據(jù)的可靠性。例如在checkpoint的過(guò)程中，分為幾個(gè)階段，通過(guò)不同的文件名來(lái)標(biāo)識(shí)當(dāng)前所處的狀態(tài)。為存儲(chǔ)失敗后進(jìn)行恢復(fù)提供了可能。
            5、元數(shù)據(jù)的一致性機(jī)制。
            • 首先從NameNode啟動(dòng)時(shí)，對(duì)每個(gè)備份目錄是否格式化、目錄元數(shù)據(jù)文件名是否正確等進(jìn)行檢查，確保元數(shù)據(jù)文件間的狀態(tài)一致性，然后選取最新的加載到內(nèi)存，這樣可以確保HDFS當(dāng)前狀態(tài)和最后一次關(guān)閉時(shí)的狀態(tài)一致性。
            • 其次，通過(guò)異常輸出流的處理，可以確保正常輸出流數(shù)據(jù)的一致性。
            • 運(yùn)用同步機(jī)制，確保了輸出流一致性問(wèn)題。

              本文轉(zhuǎn)自：              http://blog.csdn.net/kntao/article/details/7770597
            
	
            posted on 2013-05-24 15:29 王海光 閱讀(486) 評(píng)論(0)  編輯 收藏 引用  所屬分類(lèi): Linux 
            

            







            
	    
    




            






            

				

            


    
    







            
	   
	



久久国产免费观看精品3|
成人久久综合网|
91精品国产91久久|
伊人久久大香线蕉综合5g|
中文字幕无码久久精品青草|
精品人妻伦九区久久AAA片69|
天堂久久天堂AV色综合|
久久综合亚洲色HEZYO国产|
A级毛片无码久久精品免费|
久久久久亚洲AV无码专区网站
|
狠狠色丁香婷婷综合久久来|
亚洲精品tv久久久久久久久久|
久久综合给合久久国产免费|
囯产极品美女高潮无套久久久|
久久91亚洲人成电影网站|
久久精品国产99国产精品亚洲|
精品国产青草久久久久福利|
久久午夜电影网|
久久婷婷久久一区二区三区|
亚洲色欲久久久综合网东京热|
久久亚洲欧洲国产综合|
麻豆精品久久久一区二区|
77777亚洲午夜久久多喷|
一本久久a久久精品亚洲|
色综合久久久久综合99|
日日躁夜夜躁狠狠久久AV|
日本久久久久久久久久|
伊人久久综合精品无码AV专区|
狠狠色综合久久久久尤物|
国产精品久久精品|
久久青草国产精品一区|
久久精品国产精品亚洲精品|
国产亚洲色婷婷久久99精品|
久久久无码一区二区三区|
波多野结衣AV无码久久一区|
欧美伊人久久大香线蕉综合|
国色天香久久久久久久小说|
久久WWW免费人成一看片|
99久久国产宗和精品1上映|
少妇久久久久久久久久|
久久99精品久久久久久动态图
|