Hadoop学习十四:Hadoop-Hdfs FSDataset源码

TiDBPingCAP 2013-05-30

一.FSDataset类图

 Hadoop学习十四:Hadoop-Hdfs FSDataset源码

二.FSVolume FSDir物理概念

 Hadoop学习十四:Hadoop-Hdfs FSDataset源码

三.Block

  1.  Block类只代表一个block的标识,看Block类的属性便知;Block类不代表block文件。
  2. blk_1150083481087817002是block;%hadoop_home%/dfs/data/current/blk_115008348108781700是block文件。
  3. block包含block blk_1150083481087817002和block元数据 blk_1150083481087817002_1007.meta。本系列博客中没有特别说明时,block只表示block blk_1150083481087817002。
//blk_1150083481087817002
//blk_1150083481087817002_1007.meta
public class Block implements Writable, Comparable<Block> {

	//change fileName to id
	static long filename2id(String name) {
		return Long.parseLong(name.substring("blk_".length()));
	}
	//change id to fileName
	public String getBlockName() {
		return "blk_" + String.valueOf(blockId);
	}

	private long blockId;			//block id:1150083481087817002
	private long numBytes;		//block大小
	private long generationStamp; //从1000L开始:1007 当两个块进行比较的时候,当它们的hashcode相同时,便用generationStamp进行比较

	public Block() {
		this(0, 0, 0);
	}

	//blockId相同
	//generationStamp时间相同两个条件
	public boolean equals(Object o) {
		if (!(o instanceof Block)) {
			return false;
		}
		final Block that = (Block) o;
		return this.blockId == that.blockId
				&& GenerationStamp.equalsWithWildcard(this.generationStamp,
						that.generationStamp);
	}

	//根据blockId计算hashcode
	public int hashCode() {
		return 37 * 17 + (int) (blockId ^ (blockId >>> 32));
	}
}

四.BlockAndFile

      BlockAndFile类代表了block与block文件的对应关系。

// block与block文件的对应关系
	static class BlockAndFile implements Comparable<BlockAndFile> {
		final Block block;
		// absolute path eg:%hadoop_home%/dfs/data/current/blk_1150083481087817002
		final File pathfile; 

		BlockAndFile(File fullpathname, Block block) {
			this.pathfile = fullpathname;
			this.block = block;
		}

		public int compareTo(BlockAndFile o) {
			return this.block.compareTo(o.block);
		}
	}

五.DatanodeBlockInfo

  1.  DatanodeBlockInfo保存了block在文件系统上的信息,包含block存放的卷(FSVolume),文件名和detach状态。
  2. detach状态:系统在升级时会创建一个snapshot,snapshot的文件和current里的数据块文件和数据块元文件是通过硬链接,指向了相同的内容。当我们需要改变current里的文件时,如果不进行detach操作,那么,修改的内容就会影响snapshot里的文件,这时,我们需要将对应的硬链接解除掉。方法很简单,就是在临时文件夹里,复制文件,然后将临时文件改名成为current里的对应文件,这样的话,current里的文件和snapshot里的文件就detach了。这样的技术,也叫copy-on-write,是一种有效提高系统性能的方法。DatanodeBlockInfo中的detachBlock,能够对Block对应的数据文件和元数据文件进行detach操作。
    //Block----->DatanodeBlockInfo
    class DatanodeBlockInfo {
    
      private FSVolume volume;       // block所在的FSVolume
      private File     file;         // block文件
      private boolean detached;      // copy-on-write done for block
    
      DatanodeBlockInfo(FSVolume vol, File file) {
        this.volume = vol;
        this.file = file;
        detached = false;
      }
    
      /**
       * 1. Copy specified file into a temporary file. 
       * 2. Then rename the temporary file to the original name. 
       * This will cause any hardlinks to the original file to be removed. 
       * The temporary files are created in the detachDir. 
       * The temporary files will be recovered (especially on Windows) on datanode restart.
       */
      private void detachFile(File file, Block b) throws IOException {
        ...
        }
      }
    
      /**
       * Returns true if this block was copied, otherwise returns false.
       */
      boolean detachBlock(Block block, int numLinks) throws IOException {
        。。。
      }
      
    }
     

六.FSDir

  1. FSDir是保存block的文件夹。
  2. FSDir是一个树状结构,最外层是%hadoop_home%/dfs/data/current。
  3. 初始化FSDir时,迭代初始化%hadoop_home%/dfs/data/current下的所有children FSDir,构成FSDir树。
  4. FSDir的重要方法
  • addBlock:向此FSDir中添加block,返回这个block对应的block文件。
  • getBlockAndFileInfo:获得此FSDir下所有BlockAndFile。
  • getVolumeMap:获得此FSDir下所有block到DatanodeBlockInfo的映射关系。
    // 保存block的文件夹
    	class FSDir {
    		File dir; // FSDir会有一个根目录,最外面的当然是/current
    		int numBlocks = 0; // FSDir下的block数量
    		FSDir children[]; // FSDir下可以继续包含FSDir
    		int lastChildIdx = 0; // 存储上一个数据块的子目录序号
    
    		// 初始化时,构建FSDir树
    		public FSDir(File dir) throws IOException {
    			this.dir = dir;
    			this.children = null;
    			File[] files = FileUtil.listFiles(dir);
    			int numChildren = 0;
    			for (int idx = 0; idx < files.length; idx++) {
    				if (files[idx].isDirectory()) {
    					numChildren++;
    				} else if (Block.isBlockFilename(files[idx])) {
    					numBlocks++;
    				}
    			}
    			if (numChildren > 0) {
    				children = new FSDir[numChildren];
    				int curdir = 0;
    				for (int idx = 0; idx < files.length; idx++) {
    					if (files[idx].isDirectory()) {
    						// 迭代初始化children FSDir
    						children[curdir] = new FSDir(files[idx]);
    						curdir++;
    					}
    				}
    			}
    		}
    
    		public File addBlock(Block b, File src) throws IOException {
    			// First try without creating subdirectories
    			File file = addBlock(b, src, false, false);
    			return (file != null) ? file : addBlock(b, src, true, true);
    		}
    
    		private File addBlock(Block b, File src, boolean createOk,
    				boolean resetIdx) throws IOException {
    			// DataNode节点会首先把文件的数据块存储到存储路径的子目录current/下
    			if (numBlocks < maxBlocksPerDir) {
    				// src:tmp下
    				// dest:current下
    				File dest = new File(dir, b.getBlockName());
    				// metaData:tmp下
    				// newmeta:current下
    				File metaData = getMetaFile(src, b);
    				File newmeta = getMetaFile(dest, b);
    				// tmp下metaData移到current下,tmp下block移到current下
    				if (!metaData.renameTo(newmeta) || !src.renameTo(dest)) {
    					throw new IOException("could not move files for " + b
    							+ " from tmp to " + dest.getAbsolutePath());
    				}
    				numBlocks += 1;
    				return dest;
    			}
    
    			// 当子目录current/中已经存储了maxBlocksPerDir个数据块之后
    			// 就会在目录current/下创建maxBlocksPerDir个子目录,然后从中选择一个子目录,把数据块存储到这个子目录中;
    			// 如果选择的子目录也已经存储了maxBlocksPerDir个数据块,则又在这个子目录下创建maxBlocksPerDir个子目录,从这些子目录中选一个来存储数据块
    			// 就这样一次递归下去,直到存储路径的剩余存储空间不够存储一个数据块为止。
    			// maxBlocksPerDir的默认值是64,但也可以通过DataNode的配置文件来设置,它对应的配置选项是dsf.datanode.numblocks。
    			if (lastChildIdx < 0 && resetIdx) {
    				// reset so that all children will be checked
    				lastChildIdx = random.nextInt(children.length);
    			}
    
    			if (lastChildIdx >= 0 && children != null) {
    				// Check if any child-tree has room for a block.
    				for (int i = 0; i < children.length; i++) {
    					int idx = (lastChildIdx + i) % children.length;
    					File file = children[idx].addBlock(b, src, false, resetIdx);
    					if (file != null) {
    						lastChildIdx = idx;
    						return file;
    					}
    				}
    				lastChildIdx = -1;
    			}
    
    			if (!createOk) {
    				return null;
    			}
    
    			if (children == null || children.length == 0) {
    				children = new FSDir[maxBlocksPerDir];
    				for (int idx = 0; idx < maxBlocksPerDir; idx++) {
    					children[idx] = new FSDir(new File(dir,
    							DataStorage.BLOCK_SUBDIR_PREFIX + idx));
    				}
    			}
    
    			// now pick a child randomly for creating a new set of subdirs.
    			lastChildIdx = random.nextInt(children.length);
    			return children[lastChildIdx].addBlock(b, src, true, false);
    		}
    
    		// 获得此FSDir下所有BlockAndFile
    		void getBlockAndFileInfo(TreeSet<BlockAndFile> blockSet) {
    			// 迭代children FSDir
    			if (children != null) {
    				for (int i = 0; i < children.length; i++) {
    					children[i].getBlockAndFileInfo(blockSet);
    				}
    			}
    
    			File blockFiles[] = dir.listFiles();
    			for (int i = 0; i < blockFiles.length; i++) {
    				if (Block.isBlockFilename(blockFiles[i])) {
    					long genStamp = FSDataset.getGenerationStampFromFile(
    							blockFiles, blockFiles[i]);
    					Block block = new Block(blockFiles[i],
    							blockFiles[i].length(), genStamp);
    					blockSet.add(new BlockAndFile(blockFiles[i]
    							.getAbsoluteFile(), block));
    				}
    			}
    		}
    
    		// 建立Block到DatanodeBlockInfo的映射关系
    		void getVolumeMap(HashMap<Block, DatanodeBlockInfo> volumeMap, FSVolume volume) {
    			// 迭代children FSDir
    			if (children != null) {
    				for (int i = 0; i < children.length; i++) {
    					children[i].getVolumeMap(volumeMap, volume);
    				}
    			}
    
    			File blockFiles[] = dir.listFiles();
    			if (blockFiles != null) {
    				for (int i = 0; i < blockFiles.length; i++) {
    					if (Block.isBlockFilename(blockFiles[i])) {
    						long genStamp = FSDataset.getGenerationStampFromFile(
    								blockFiles, blockFiles[i]);
    						volumeMap.put(
    								new Block(blockFiles[i],
    										blockFiles[i].length(), genStamp),
    								new DatanodeBlockInfo(volume, blockFiles[i]));
    					}
    				}
    			}
    		}
    		
    	}
     

七.FSVolume

  1.  FSVolume对应着DataNode上的一个Storage。一个DataNode可以配置多个Storage,一个DataNode包含多个FSVolume。
  2. FSVolume的重要方法
  • getDfsUsed磁盘使用量 getCapacity磁盘大小  getAvailable磁盘可用量
  • addBlock:向FSVolume中添加block,调用FSDir.addBlock完成。
  • getVolumeMap:获得此FSVolume下所有block到DatanodeBlockInfo的映射关系,调用FSDir.getVolumeMap完成。 
    // FSVolume对应一个Storage
    	// 一个DataNode可以配置多个Storage,一个DataNode包含多个FSVolume
    	class FSVolume {
    		private File currentDir;
    		private FSDir dataDir;
    		private File tmpDir;
    		private File blocksBeingWritten; // clients write here
    		private File detachDir; // copy on write for blocks in snapshot
    		private DF usage;
    		private DU dfsUsage;
    		//<property>   
    		//  <name>dfs.datanode.du.reserved</name>   
    		//	<value>1024</value>   
    		//</property>  
    		//每个磁盘写入点能预留1K的空间来
    		private long reserved;
    
    		// 初始化一个FSVolume
    		FSVolume(File currentDir, Configuration conf) throws IOException {
    			this.reserved = conf.getLong("dfs.datanode.du.reserved", 0);
    			this.dataDir = new FSDir(currentDir);
    			this.currentDir = currentDir;
    			//根据parent初始化下面各属性,parent is %hadoop_home%/dfs/data
    			File parent = currentDir.getParentFile();
    
    			this.detachDir = new File(parent, "detach");
    
    			// remove all blocks from "tmp" directory. These were either created
    			// by pre-append clients (0.18.x) or are part of replication
    			// request.
    			// They can be safely removed.
    			this.tmpDir = new File(parent, "tmp");
    			if (tmpDir.exists()) {
    				FileUtil.fullyDelete(tmpDir);
    			}
    
    			// Files that were being written when the datanode was last shutdown
    			// should not be deleted.
    			blocksBeingWritten = new File(parent, "blocksBeingWritten");
    			...
    			
    			this.usage = new DF(parent, conf);
    			this.dfsUsage = new DU(parent, conf);
    			this.dfsUsage.start();
    		}
    
    		//getDfsUsed getCapacity  getAvailable
    		long get*() throws IOException {
    			return dfsUsage.get*();
    		}
    
    		File addBlock(Block b, File f) throws IOException {
    			//调用FSDir的addBlock
    			File blockFile = dataDir.addBlock(b, f);
    			File metaFile = getMetaFile(blockFile, b);
    			// add 后,磁盘使用量增加
    			dfsUsage.incDfsUsed(b.getNumBytes() + metaFile.length());
    			return blockFile;
    		}
    
    
    		//当前FSVolume下的volumeMap
    		void getVolumeMap(HashMap<Block, DatanodeBlockInfo> volumeMap) {
    			dataDir.getVolumeMap(volumeMap, this);
    		}
    
    	}
     

八.DF DU

  1.  DF被设计用来获取dirPath路径所在的磁盘的空间状态信息,对应的unix的shell脚本命令格式是:df -k path。
  2. DU类实现了unix的du命令,显示文件或目录dirPath占用磁盘空间的大小信息。
    public class DF extends Shell {
    
      /** Default DF refresh interval. */
      public static final long DF_INTERVAL_DEFAULT = 3 * 1000;
      
      private final String dirPath;	//执行df命令所在工作目录	
      private final File dirFile;	//执行df命令所在工作目录文件夹	
      private String filesystem;	//磁盘设备名   
      private String mount;	//磁盘挂载位置 
      
      //初始化dirPath and dirFile
      public DF(File path, long dfInterval) throws IOException {
        super(dfInterval);
        this.dirPath = path.getCanonicalPath();
        this.dirFile = path.getCanonicalFile();
      }
      
    
      //getCapacity getUsed getAvailable
      public long get*() {
        return dirFile.get*();
      }
    }
    
    
    public class DU extends Shell {
      private String  dirPath;		//执行du命令所在工作目录	
    }
     

九.FSVolumeSet

  1.  管理一个DataNode下所有的FSVolume。
  2. FSVolume的重要方法
  • getVolumeMap:获得FSVolume[]下所有block到DatanodeBlockInfo的映射关系,叠加FSVolume.getVolumeMap实现。 
  • getDfsUsed磁盘使用量 getCapacity磁盘大小  getRemaining磁盘可用量,叠加FSVolume.x实现。
    //管理一个DataNode下所有的FSVolume
    	static class FSVolumeSet {
    		FSVolume[] volumes = null;
    		int curVolume = 0;
    
    		FSVolumeSet(FSVolume[] volumes) {
    			this.volumes = volumes;
    		}
    
    		//向DataNode添加block时,根据blockSize,获取第一个大于blockSize的FSVolume
    		synchronized FSVolume getNextVolume(long blockSize) throws IOException {
    
    			// make sure we are not out of bounds
    			if (curVolume >= volumes.length) {
    				curVolume = 0;
    			}
    
    			int startVolume = curVolume;
    
    			while (true) {
    				FSVolume volume = volumes[curVolume];
    				curVolume = (curVolume + 1) % volumes.length;
    				if (volume.getAvailable() > blockSize) {
    					return volume;
    				}
    				//空间不足
    				if (curVolume == startVolume) {
    					throw new DiskOutOfSpaceException("Insufficient space for an additional block");
    				}
    			}
    		}
    
    		long get*() throws IOException {
    			叠加每个FSVolume
    		}
    
    		//所有FSVolume下的volumeMap
    		synchronized void getVolumeMap(HashMap<Block, DatanodeBlockInfo> volumeMap) {
    			for (int idx = 0; idx < volumes.length; idx++) {
    				volumes[idx].getVolumeMap(volumeMap);
    			}
    		}
    	}

十.ActiveFile

     ActiveFile对象保存了一个文件,和操作这个文件的线程,线程有可能有多个。

static class ActiveFile {
		final File file;
		final List<Thread> threads = new ArrayList<Thread>(2);

		//初始化ActiveFile时会自动地把当前线程加入其中
		ActiveFile(File f, List<Thread> list) {
			this(f, false);
			if (list != null) {
				threads.addAll(list);
			}
			threads.add(Thread.currentThread());
		}
	}

十一.FSDataset

  1. FSDataset manages a set of data blocks.通过FSVolumeSet 管理。
  2. FSDataset实现了FSDatasetInterface接口,FSDatasetInterface接口是DataNode对底层存储的抽象。
    public class FSDataset implements FSConstants, FSDatasetInterface {
    	
    	//所有FSVolume
    	FSVolumeSet volumes;
    	//所有Block到DatanodeBlockInfo的映射
    	HashMap<Block, DatanodeBlockInfo> volumeMap = new HashMap<Block, DatanodeBlockInfo>();;
    	//所有Block到ActiveFile的映射,也就是说,说有正在创建的Block,都会记录在ongoingCreates里。
    	private HashMap<Block, ActiveFile> ongoingCreates = new HashMap<Block, ActiveFile>();
    
    	//初始化FSDataset时初始化volumes and volumeMap
    	public FSDataset(DataStorage storage, Configuration conf)throws IOException {
    		FSVolume[] volArray = new FSVolume[storage.getNumStorageDirs()];
    		for (int idx = 0; idx < storage.getNumStorageDirs(); idx++) {
    			volArray[idx] = new FSVolume(storage.getStorageDir(idx)
    					.getCurrentDir(), conf);
    		}
    		volumes = new FSVolumeSet(volArray);
    		volumes.getVolumeMap(volumeMap);
    	}
    	
    	//=================================== 根据block 的几个方法 开始===================================
    	//得到block文件
    	public synchronized File getBlockFile(Block b) throws IOException ;
    	//得到block元文件
    	protected File getMetaFile(Block b) throws IOException  ;
    	//得到block的元数据长度。
    	public long getMetaDataLength(Block b) throws IOException  ;
    	
    	//得到InputStream MetaDataInputStream包含block长度
    	public MetaDataInputStream getMetaDataInputStream(Block b) throws IOException;
    	//得到block对应元数据文件的inputstream
    	public InputStream getBlockInputStream(Block b) throws IOException
    	//获得block对应元数据文件的inputstream, 从指定位置开始读
    	public InputStream getBlockInputStream(Block b, long seekOffset) throws IOException;
    	//得到Block的临时输入流。注意,临时输入流是指对应的文件处于tmp目录中。
    	//新创建块时,块数据应该写在tmp目录中,直到写操作成功,文件才会被移动到current目录中,如果失败,就不会影响current目录了。简单方法。
    	public BlockInputStreams getTmpInputStreams(Block b, long blkoff, long ckoff) throws IOException;
    	
    	//得到一个block的输出流。BlockWriteStreams既包含了数据输出流,也包含了元数据(校验文件)输出流。
    	//参数isRecovery说明这次写是不是对以前失败的写的一次恢复操作。
    	//正常的写操作流程:首先,如果输入的block是个正常的数据块,或当前的block已经有线程在写,writeToBlock会抛出一个异常。
    	//否则,将创建相应的临时数据文件和临时元数据文件,并把相关信息,创建一个ActiveFile对象,记录到ongoingCreates中,并创建返回的BlockWriteStreams。
    	//前面我们已经提过,建立新的ActiveFile时,当前线程会自动保存在ActiveFile的threads中。
    	//以blk_3148782637964391313为例,
    	//当DataNode需要为Block ID为3148782637964391313创建写流时,DataNode创建文件tmp/blk_3148782637964391313做为临时数据文件,
    	//对应的meta文件是tmp/blk_3148782637964391313_XXXXXX.meta。其中XXXXXX是版本号。
    	//isRecovery为true时,表明我们需要从某一次不成功的写中恢复,流程相对于正常流程复杂。
    	//如果不成功的写是由于提交(参考finalizeBlock方法)后的确认信息没有收到,先创建一个detached文件(备份)。
    	//接着,writeToBlock检查是否有还有对文件写的线程,如果有,则通过线程的interrupt方法,强制结束线程。这就是说,如果有线程还在写对应的文件块,该线程将被终止。
    	//同时,从ongoingCreates中移除对应的信息。接下来将根据临时文件是否存在,创建/复用临时数据文件和临时数据元文件。
    	//后续操作就和正常流程一样,根据相关信息,创建一个ActiveFile对象,记录到ongoingCreates中
    	public BlockWriteStreams writeToBlock(Block b, boolean isRecovery, boolean isReplicationRequest) throws IOException;
    	//提交(或叫:结束finalize)通过writeToBlock打开的block,这意味着写过程没有出错,可以正式把Block从tmp文件夹放到current文件夹。
    	//将从ongoingCreates中删除对应的block,同时将block对应的DatanodeBlockInfo,放入volumeMap中。
    	//以blk_3148782637964391313为例,当DataNode提交Block ID为3148782637964391313数据块文件时,DataNode将把tmp/blk_3148782637964391313移到current下某一个目录,
    	//以subdir12为例,这是tmp/blk_3148782637964391313将会挪到current/subdir12/blk_3148782637964391313。对应的meta文件也在目录current/subdir12下。
    	public void finalizeBlock(Block b) throws IOException;
    	//更新一个block。
    	//updateBlock的最外层是一个死循环,循环的结束条件,是没有任何和这个数据块相关的写线程。
    	//每次循环,updateBlock都会去调用一个叫tryUpdateBlock的内部方法。
    	//tryUpdateBlock发现已经没有线程在写这个块,就会跟新和这个数据块相关的信息,包括元文件和内存中的映射表volumeMap。
    	//如果tryUpdateBlock发现还有活跃的线程和该块关联,那么,updateBlock会试图结束该线程,并等在join上等待。
    	public void updateBlock(Block oldblock, Block newblock) throws IOException;
    	//取消通过writeToBlock打开的block,与finalizeBlock方法作用相反。
    	public void unfinalizeBlock(Block b) throws IOException;
    	//=================================== 根据block 的几个方法 结束===================================
    	
    	//getDfsUsed getCapacity  getRemaining
    	public long get*() throws IOException {
    		return volumes.get*();
    	}
    
    }
     

相关推荐