Hadoop-0.20.0源代码分析（07）
前面分析了与操作系统有关的Shell命令，它们用于与操作系统进行命令行方式的交互。在Hadoop中，自定义了FileSystem文件系统，这是基于Unix操作系统之上的文件系统，为了方便对FileSystem的管理，通过org.apache.hadoop.fs.FsShell类定义了对Hadoop FileSystem文件系统进行命令行方式管理的命令实现。先给出对Hadoop文件系统进行管理的命令实现类的继承层次关系：
◦org.apache.hadoop.conf.Configured(implements org.apache.hadoop.conf.Configurable)◦org.apache.hadoop.fs.FsShell(implements org.apache.hadoop.util.Tool)◦org.apache.hadoop.hdfs.tools.DFSAdmin
由于DFSAdmin类是对HDFS分布式文件系统提供基于命令行的管理功能，这里先不对DFSAdmin进行分析，在后面分析HDFS实现的时候，进行详细分析理解。
Configured就不用多说了，是Hadoop配置类的高层抽象。
Tool接口支持命令行方式的处理，如果需要通过命令行方式来执行一定的任务，都可以实现该接口，通过该接口定义的run方法来运行命令行。由于它继承自Configurable 接口，使得实现Tool的接口可以对特定的待执行的任务进行详细配置，满足执行一个命令能够完成任务的要求。下面是接口的定义：
public interface Tool extends Configurable {int run(String [] args) throws Exception;}
在Hadoop中，Tool接口主要是为进行MapReduce并行计算而定义的，这里FsShell类实现了该接口，其实也是使得命令行执行与任务关联起来，通过执行命令行，而执行设置的待完成的任务。
下面来看FsShell类的具体实现。
既然，FsShell是与命令行有关的，那么我们就从其中对指定的命令实现的角度来看，分别对每个命令的实现进行阅读分析。在分析每个命令实现过程之前，先看一下该类中printUsage方法的执行，该方法能够打印出全部命令用法的信息，如下所示：
Usage: java FsShell[-ls <path>][-lsr <path>][-du <path>][-dus <path>][-count[-q] <path>][-mv <src> <dst>][-cp <src> <dst>][-rm <path>][-rmr <path>][-expunge][-put <localsrc> ... <dst>][-copyFromLocal <localsrc> ... <dst>][-moveFromLocal <localsrc> ... <dst>][-get [-ignoreCrc] [-crc] <src> <localdst>][-getmerge <src> <localdst> ][-cat <src>][-text <src>][-copyToLocal [-ignoreCrc] [-crc] <src> <localdst>][-moveToLocal [-crc] <src> <localdst>][-mkdir <path>][-setrep [-R] [-w] <rep> <path/file>][-touchz <path>][-test - <path>][-stat <path>][-tail [-f] <file>][-chmod [-R] <MODE[,MODE]... | OCTALMODE> PATH...][-chown [-R] [:] PATH...][-chgrp [-R] GROUP PATH...][-help ]Generic options supported are-conf <configuration file> specify an application configuration file-D <property=value> use value for given property-fs <local|namenode:port> specify a namenode-jt <local|jobtracker:port> specify a job tracker-files <comma separated list of files> specify comma separated files to be copied to the map reduce cluster-libjars <comma separated list of jars> specify comma separated jar files to include in the classpath.-archives <comma separated list of archives> specify comma separated archives to be unarchived on the compute machines.The general command line syntax isbin/hadoop command
非常清晰明了，FsShell所支持的命令行，及其该命令的可以设置的参数，都在上述列表中显示出来。
另外，对于每个命令的帮助信息，都可以通过printHelp方法得到，例如，如果想要得到命令“ls”的帮助信息，调用printHelp("ls");即可。如果想要得到全部命令的帮助信息，只要给printHelp随便传入一个非命令字符串，如printHelp("hashyes3532333");，将打印出全部命令帮助信息，下面是一个帮助信息的片段：
hadoop fs is the command to execute fs commands. The full syntax is: hadoop fs [-fs <local | file system URI>] [-conf <configuration file>][-D <property=value>] [-ls <path>] [-lsr <path>] [-du <path>][-dus <path>] [-mv <src> <dst>] [-cp <src> <dst>] [-rm <src>][-rmr <src>] [-put <localsrc> ... <dst>] [-copyFromLocal <localsrc> ... <dst>][-moveFromLocal <localsrc> ... <dst>] [-get [-ignoreCrc] [-crc] <src> <localdst>[-getmerge <src> <localdst> ] [-cat <src>][-copyToLocal [-ignoreCrc] [-crc] <src> <localdst>] [-moveToLocal <src> <localdst>][-mkdir <path>] [-report] [-setrep [-R] [-w] <rep> <path/file>][-touchz <path>] [-test - <path>] [-stat <path>][-tail [-f] <path>] [-text <path>][-chmod [-R] <MODE[,MODE]... | OCTALMODE> PATH...][-chown [-R] [:] PATH...][-chgrp [-R] GROUP PATH...][-count[-q] <path>][-help ]-fs : Specify the file system to use.If not specified, the current configuration is used, taken from the following, in increasing precedence: core-default.xml inside the hadoop jar file core-site.xml in $HADOOP_CONF_DIR 'local' means use the local file system as your DFS. <file system URI> specifies a particular file system to contact. This argument is optional but if used must appearappear first on the command line.Exactly one additionalargument must be specified.
下面介绍每个命令的实现：
[*]ls与lsr命令
执行ls命令，能够列出匹配指定Path下的全部文件，并且不递归列出子目录中文件；lsr能够列出指定Path下的所有文件，并且如果存在子目录，也会递归列出子目录中的文件。实现这两个命令的方法均为ls方法，如下所示：
/*** 列出满足模式srcf的全部文件* @param srcf 文件模式* @param recursive 是否递归列出*/private int ls(String srcf, boolean recursive) throws IOException {Path srcPath = new Path(srcf);FileSystem srcFs = srcPath.getFileSystem(this.getConf()); // 通过构造Path类实例，获取它所属的FileSystem文件系统FileStatus[] srcs = srcFs.globStatus(srcPath); // 获取到文件系统srcFs中匹配srcPath模式的全部按照文件名称排好序的文件（不包括校验和文件），每个文件对应一个FileStatusif (srcs==null || srcs.length==0) {throw new FileNotFoundException("Cannot access " + srcf +": No such file or directory.");}boolean printHeader = (srcs.length == 1) ? true: false; // 两种情况：如果获取到一个文件FileStatus，表示只有一个目录或者文件，需要打印出列表头部信息；否则返回多个FileStatus，需要循环并递归遍历，不打印出列表头部信息int numOfErrors = 0;for(int i=0; i<srcs.length; i++) {numOfErrors += ls(srcs, srcFs, recursive, printHeader); // 调用ls，递归列出文件}return numOfErrors == 0 ? 0 : -1;}
实际上，执行ls命令真正实现执行的过程在重载的另一个方法ls中，如下所示：
private int ls(FileStatus src, FileSystem srcFs, boolean recursive, boolean printHeader) throws IOException {final String cmd = recursive? "lsr": "ls"; // 根据recursive判断，是否递归列出文件，如果是则命令名称为lsr，否则命令为lsfinal FileStatus[] items = shellListStatus(cmd, srcFs, src); // 调用shellListStatus方法，文件系统从srcFs中获取src中的全部FileStatus[]（如果src是文件而非目录，直接返回它自身）if (items == null) {return 1;} else {int numOfErrors = 0;if (!recursive && printHeader) { // 如果指定不递归列出if (items.length != 0) {System.out.println("Found " + items.length + " items");}}int maxReplication = 3, maxLen = 10, maxOwner = 0,maxGroup = 0; // for(int i = 0; i < items.length; i++) {FileStatus stat = items;int replication = String.valueOf(stat.getReplication()).length(); // stat对应文件的replication因子int len = String.valueOf(stat.getLen()).length(); // stat对应的文件的长度int owner = String.valueOf(stat.getOwner()).length(); // stat对应的文件的属主数int group = String.valueOf(stat.getGroup()).length(); // stat对应的文件属组数if (replication > maxReplication) maxReplication = replication; // 有可能一个文件的副本数超过指定的最大replication因子值if (len > maxLen) maxLen = len; // 超过文件最大长度的情况if (owner > maxOwner)maxOwner = owner; // 超过最大属主数 if (group > maxGroup)maxGroup = group; // 超过最大属组数}for (int i = 0; i < items.length; i++) {FileStatus stat = items;Path cur = stat.getPath();String mdate = dateForm.format(new Date(stat.getModificationTime())); // 格式化stat对应的文件的修改时间 System.out.print((stat.isDir() ? "d" : "-") +stat.getPermission() + " "); // 输出stat对应文件权限信息：若是目录就以"d权限"格式输出，若是文件则以"-权限"格式输出System.out.printf("%"+ maxReplication + "s ", (!stat.isDir() ? stat.getReplication() : "-")); // 输出stat对应文件最replication因子信息到输出流中if (maxOwner > 0)System.out.printf("%-"+ maxOwner + "s ", stat.getOwner()); // 输出stat对应文件的属主数到输出流中if (maxGroup > 0)System.out.printf("%-"+ maxGroup + "s ", stat.getGroup()); // 输出stat对应文件的属组数到输出流中System.out.printf("%"+ maxLen + "d ", stat.getLen()); // 输出stat对应文件的最大长度信息到输出流中System.out.print(mdate + " "); // 输出格式化的文件修改时间System.out.println(cur.toUri().getPath()); // 输出stat对应文件的路径信息if (recursive && stat.isDir()) { // 如果stat对应的是目录，而且要求递归列出numOfErrors += ls(stat,srcFs, recursive, printHeader); // 递归调用ls}}return numOfErrors;}}
lsr命令是递归列出满足给定模式的全部文件，也是基于上述方法实现的。
通过上面的ls的实现可知，列出FileSystem文件系统中的数据，是通过获取到该文件系统中保存的文件的FileStatus实例，因为FileStatus描述了位于该文件系统中对应文件的详细信息，然后通过它来打印出文件类表（包含必要的文件属性信息）。
[*]du与dus命令
du命令列出满足给定模式的全部文件对应的长度信息，dus执行后列出了满足给定模式的每个文件或目录的磁盘使用情况摘要信息，比du命令执行得到的结果信息要详细。
du命令实现是通过du方法，如下：
void du(String src) throws IOException {Path srcPath = new Path(src);FileSystem srcFs = srcPath.getFileSystem(getConf()); // 获取到Path对应的FileSystem文件系统Path[] pathItems = FileUtil.stat2Paths(srcFs.globStatus(srcPath), srcPath); // 调用：将从srcFs文件系统中获取到经过srcPath过滤的FileStatus[]转换为Path数组FileStatus items[] = srcFs.listStatus(pathItems); // 根据得到的满足过滤条件的Path得到对应的FileStatusif ((items == null) || ((items.length == 0) && (!srcFs.exists(srcPath)))){throw new FileNotFoundException("Cannot access " + src + ": No such file or directory.");} else {System.out.println("Found " + items.length + " items");int maxLength = 10;long length[] = new long; // length数组用来保存每个文件对应的长度信息for (int i = 0; i < items.length; i++) {length = items.isDir() ?srcFs.getContentSummary(items.getPath()).getLength() : items.getLen(); // 若items对应文件是目录，通过srcFs获取到其内容摘要信息的长度，若是普通文件，则得到其长度int len = String.valueOf(length).length();if (len > maxLength) maxLength = len;}for(int i = 0; i < items.length; i++) { // 循环遍历System.out.printf("%-"+ (maxLength + BORDER) +"d", length); // 将每个文件或者目录的长度信息写入到流中System.out.println(items.getPath());}}}
获取文件信息的方式，基本上都是一致的，通过文件系统来得到对应文件的统计信息。dus命令实现通过dus方法，与上面的实现基本类似，与ds实现不同的是，从FileSystem文件系统中获取到的文件不管是目录还行普通文件，都获取到其摘要信息（对应org.apache.hadoop.fs.ContentSummary）的长度，最后返回执行结果。
[*]mkdir命令
该命令根据跟定的字符串，创建该字符串标识的目录，实现方法为mkdir方法，实现比较简单易懂：
void mkdir(String src) throws IOException {Path f = new Path(src);FileSystem srcFs = f.getFileSystem(getConf());FileStatus fstatus = null;try {fstatus = srcFs.getFileStatus(f);if (fstatus.isDir()) { // 只有给定的src在文件系统中不存在时，才可以创建throw new IOException("cannot create directory " + src + ": File exists");}else {throw new IOException(src + " exists but " + "is not a directory");}} catch(FileNotFoundException e) {if (!srcFs.mkdirs(f)) { // 通过调用文件系统srcFs的创建目录方法，执行目录的创建throw new IOException("failed to create " + src);}}}
[*]touchz命令
该命令创建一个空文件，大小为0，通过touchz方法实现，实现的原理也是，通过调用文件系统的create方法执行文件的创建，如下所示：
void touchz(String src) throws IOException {Path f = new Path(src);FileSystem srcFs = f.getFileSystem(getConf());FileStatus st;if (srcFs.exists(f)) {st = srcFs.getFileStatus(f); // 尝试，是否能够从文件系统srcFs中获取到待创建文件的信息if (st.isDir()) { // 如果该文件时一个目录，不能创建throw new IOException(src + " is a directory");} else if (st.getLen() != 0) // 如果该文件存在，并且文件不空，也不能创建throw new IOException(src + " must be a zero-length file");}FSDataOutputStream out = srcFs.create(f); // 调用文件系统srcFs的create方法创建0长度新文件out.close();}
[*]mv命令
该命令是移动文件，并支持文件的重命名，在FsShell类中通过rename方法实现的。方法实现如下所示：
private int rename(String argv[], Configuration conf) throws IOException {int i = 0;int exitCode = 0;String cmd = argv;// 提取出命令名称String dest = argv; //命令行中最后一个参数// 如果命令行指定了大于3个参数，最后一个一定是一个目录if (argv.length > 3) {Path dst = new Path(dest); // 创建目录FileSystem dstFs = dst.getFileSystem(getConf()); // 得到该目录所在的文件系统dstFsif (!dstFs.isDirectory(dst)) { // 如果文件系统dstFs中存在dst，而且它不是一个目录，出错throw new IOException("When moving multiple files, " + "destination " + dest + " should be a directory.");}}// 循环遍历多个输入源文件，也就是在命令名称与最后一个参数之间的参数字符串for (; i < argv.length - 1; i++) {try {rename(argv, dest); // 调用：将每个源文件argv移动到dest目录中} catch (RemoteException e) {// 移动文件过程中发生异常，由hadoop server返回，打印出错信息的第一行exitCode = -1;try {String[] content;content = e.getLocalizedMessage().split("/n");System.err.println(cmd.substring(1) + ": " + content);} catch (Exception ex) {System.err.println(cmd.substring(1) + ": " + ex.getLocalizedMessage());}} catch (IOException e) {// 捕获异常exitCode = -1;System.err.println(cmd.substring(1) + ": " + e.getLocalizedMessage());}}return exitCode;}
接着，看一下上面调用的一个重载的rename方法，将一个文件进行移动和重命名操作：
void rename(String srcf, String dstf) throws IOException {Path srcPath = new Path(srcf); // 源文件Path dstPath = new Path(dstf); // 目的文件FileSystem srcFs = srcPath.getFileSystem(getConf()); // 源文件系统FileSystem dstFs = dstPath.getFileSystem(getConf()); // 目的文件系统URI srcURI = srcFs.getUri(); // 源文件系统URIURI dstURI = dstFs.getUri(); // 目的文件系统URIif (srcURI.compareTo(dstURI) != 0) { // 文件移动只支持在同一个FileSystem文件系统之上进行throw new IOException("src and destination filesystems do not match.");}Path[] srcs = FileUtil.stat2Paths(srcFs.globStatus(srcPath), srcPath); // 得到全部满足的PathPath dst = new Path(dstf);if (srcs.length > 1 && !srcFs.isDirectory(dst)) { // 输入源文件大于1个，如果目的文件不是目录，出错throw new IOException("When moving multiple files, " + "destination should be a directory.");}for(int i=0; i<srcs.length; i++) { // 迭代输入源文件if (!srcFs.rename(srcs, dst)) { // 调用文件系统srcFs的rename方法实现移动文件并重命名FileStatus srcFstatus = null;FileStatus dstFstatus = null;try {srcFstatus = srcFs.getFileStatus(srcs);} catch(FileNotFoundException e) {throw new FileNotFoundException(srcs + ": No such file or directory");}try {dstFstatus = dstFs.getFileStatus(dst);} catch(IOException e) {}if((srcFstatus!= null) && (dstFstatus!= null)) { // 移动文件失败，捕获：输入源文件为目录，目的文件不是目录if (srcFstatus.isDir()&& !dstFstatus.isDir()) {throw new IOException("cannot overwrite non directory " + dst + " with directory " + srcs);}}throw new IOException("Failed to rename " + srcs + " to " + dst);}}}
可以看到，在FsShell类中定义的mv操作，不支持在不同的FileSystem文件系统之间进行文件的移动操作。
[*]rm与rmr命令
rm命令是删除文件，rmr是递归删除给定目录的子目录中的 文件，实现方式和ls与lsr类似。也存在两个重载的delete方法实现rm与rmr命令，先看其中一个：
void delete(String srcf, final boolean recursive) throws IOException {Path srcPattern = new Path(srcf); // 根据给定的srcf模式，构造一个Pathnew DelayedExceptionThrowing() { // 延迟抛出执行该删除命令发生的异常信息@Overridevoid process(Path p, FileSystem srcFs) throws IOException {delete(p, srcFs, recursive); // 调用重载的delete方法，执行删除操作}}.globAndProcess(srcPattern, srcPattern.getFileSystem(getConf())); // 收集异常信息，以待命令执行完成后一起抛出}
上面，实际上在org.apache.hadoop.fs.FsShell.DelayedExceptionThrowing类中定义的globAndProcess方法中，循环执行了重载的delete方法，也就是真正真正实现删除的delete方法。也就是说，每调用执行delete方法，能够删除一个指定的文件。该方法实现如下所示：
private void delete(Path src, FileSystem srcFs, boolean recursive) throws IOException {if (srcFs.isDirectory(src) && !recursive) { // src是目录，且指定不进行递归删除，报错throw new IOException("Cannot remove directory /"" + src + "/", use -rmr instead");}Trash trashTmp = new Trash(srcFs, getConf()); // 构造一个回收站if (trashTmp.moveToTrash(src)) { // 将src移动到回收站中（可能是文件或者目录）System.out.println("Moved to trash: " + src);return;}if (srcFs.delete(src, true)) { // 从文件系统srcFs中删除文件srcSystem.out.println("Deleted " + src);} else {if (!srcFs.exists(src)) { // 若删除失败，查找失败原因throw new FileNotFoundException("cannot remove " + src + ": No such file or directory.");}throw new IOException("Delete failed " + src);}}
执行删除文件操作的时候，是将存在于FileSystem文件系统上的文件移动到Hadoop定义的回收站.Trash目录中。
[*]cat命令
该命令取出全部满足给定模式的文件，并缓冲到标准输出流上。
该命令实现的方法为cat方法，如下所示：
void cat(String src, boolean verifyChecksum) throws IOException {Path srcPattern = new Path(src);new DelayedExceptionThrowing() { // 延迟抛出执行命令捕获到的异常信息@Overridevoid process(Path p, FileSystem srcFs) throws IOException {if (srcFs.getFileStatus(p).isDir()) {throw new IOException("Source must be a file.");}printToStdout(srcFs.open(p)); // 调用：执行命令}}.globAndProcess(srcPattern, getSrcFileSystem(srcPattern, verifyChecksum));}
调用方法printToStdout真正执行命令，该方法实现如下所示：
private void printToStdout(InputStream in) throws IOException {try {IOUtils.copyBytes(in, System.out, getConf(), false); // 使用IOUtils工具类将in流拷贝到System.out流中} finally {in.close();}}
可以查阅IOUtils类中的具体实现。这里，先不对拷贝的具体实现进行分析，在后面会单独对涉及拷贝操作的实现进行详细分析。
[*]stat命令
该命令可以得到一个文件的详细统计信息，实现方法为stat方法，实现比较简单，不再累述。
[*]tail命令
tail命令执行显示一个文件的最后1KB内容，在tail方法中实现，如下所示：
private void tail(String[] cmd, int pos) throws IOException {CommandFormat c = new CommandFormat("tail", 1, 1, "f"); // 构造一个解析命令行参数的CommandFormat对象String src = null;Path path = null;try {List<String> parameters = c.parse(cmd, pos); // 解析cmd的参数src = parameters.get(0); // 文件参数} catch(IllegalArgumentException iae) {System.err.println("Usage: java FsShell " + TAIL_USAGE);throw iae;}boolean foption = c.getOpt("f") ? true: false; // 判断是否设置了-f选项path = new Path(src);FileSystem srcFs = path.getFileSystem(getConf()); // 获取到Path对应的文件系统if (srcFs.isDirectory(path)) { // 若path是目录，出错throw new IOException("Source must be a file.");}long fileSize = srcFs.getFileStatus(path).getLen(); // 计算path文件的长度long offset = (fileSize > 1024) ? fileSize - 1024: 0; // 计算开始的偏移位置while (true) {FSDataInputStream in = srcFs.open(path); // 打开文件in.seek(offset); // 定位到offset位置IOUtils.copyBytes(in, System.out, 1024, false); // 将输入流in拷贝到Syste.out标准输出流中offset = in.getPos(); // 重新设置开始偏移位置in.close(); // 关闭输入流inif (!foption) { // 如果没有设置-f选项，直接退出break;}fileSize = srcFs.getFileStatus(path).getLen(); // 设置了-f选项，显示向文件path追加写入数据的起始位置offset = (fileSize > offset) ? offset: fileSize;try {Thread.sleep(5000);} catch (InterruptedException e) {break;}}}
[*]setrep命令
该命令是设置满足给定模式的文件的副本因子（replication factor）。不仅可以通过该类实现的setReplication方法对单个文件设置副本因子，也可以递归设置某个目录的所有文件的副本因子。实现设置副本因子的方法在该类中有多个，包括重载的方法，先队下面的方法来分析：
private void setReplication(String[] cmd, int pos) throws IOException {CommandFormat c = new CommandFormat("setrep", 2, 2, "R", "w"); // 解析命令行String dst = null;short rep = 0; // 初始化副本因子try {List<String> parameters = c.parse(cmd, pos); // 从位置pos出开始，解析出命令行中的全部参数列表rep = Short.parseShort(parameters.get(0)); // 第一个参数就是副本因子的值dst = parameters.get(1); // 第二个参数是带设置副本因子的文件} catch (NumberFormatException nfe) {System.err.println("Illegal replication, a positive integer expected");throw nfe;}catch(IllegalArgumentException iae) {System.err.println("Usage: java FsShell " + SETREP_SHORT_USAGE);throw iae;}if (rep < 1) { // 不能将副本因子设置为负数System.err.println("Cannot set replication to: " + rep);throw new IllegalArgumentException("replication must be >= 1");}List<Path> waitList = c.getOpt("w")? new ArrayList<Path>(): null; // 如果设置了-w选项，会将待设置副本因子完成的文件Path暂时缓存到列表ArrayList中setReplication(rep, dst, c.getOpt("R"), waitList); // 调用重载的setReplication方法，设置副本因子if (waitList != null) {waitForReplication(waitList, rep); // 更新waitList中文件的块的副本因子信息}}
看一下重载的setReplication方法设置副本因子的实现过程：
void setReplication(short newRep, String srcf, boolean recursive, List<Path> waitingList) throws IOException {Path srcPath = new Path(srcf);FileSystem srcFs = srcPath.getFileSystem(getConf()); // 获取到srcf所在的文件系统srcFsPath[] srcs = FileUtil.stat2Paths(srcFs.globStatus(srcPath), srcPath); // 得到满足srcf模式的全部Path文件for(int i=0; i<srcs.length; i++) { // 对每一个Path数组srcs中的每一个文件Path设置副本因子setReplication(newRep, srcFs, srcs, recursive, waitingList); // 调用另一个重载的setReplication方法进行副本因子的设置}}
继续看上面方法调用的setReplication方法，实现如下所示：
/*** 该方法递归设置每个src及其子目录中文件的副本因子*/private void setReplication(short newRep, FileSystem srcFs, Path src, boolean recursive, List<Path> waitingList) throws IOException {if (!srcFs.getFileStatus(src).isDir()) { // 递归出口：如果src是一个普通文件（而非目录）setFileReplication(src, srcFs, newRep, waitingList); // 调用setFileReplication方法设置文件src的副本因子return; }FileStatus items[] = srcFs.listStatus(src); // 如果src是目录，获取该目录中所有的文件FileStatus数组if (items == null) {throw new IOException("Could not get listing for " + src);} else {for (int i = 0; i < items.length; i++) { // 分治思想：对每一个items进行递归设置副本因子if (!items.isDir()) { setFileReplication(items.getPath(), srcFs, newRep, waitingList); // items不是目录，调用setFileReplication方法设置副本因子} else if (recursive) { // 如果指定recursive=true，且items是一个目录setReplication(newRep, srcFs, items.getPath(), recursive, waitingList); // 递归设置副本因子}}}}
上面方法调用了setFileReplication方法，设置一个非目录文件的副本因子，实现过程如下所示：
private void setFileReplication(Path file, FileSystem srcFs, short newRep, List<Path> waitList) throws IOException {if (srcFs.setReplication(file, newRep)) { // 调用文件系统srcFs的设置副本因子的方法，设置副本因子if (waitList != null) {waitList.add(file); // 将设置副本因子完成的文件file加入到waitList列表}System.out.println("Replication " + newRep + " set: " + file);} else {System.err.println("Could not set replication for: " + file);}}
我们再回到最前面重载的setReplication方法，已经完成了设置副本因子的任务，然后需要执行waitForReplication(waitList, rep);语句。此时，全部需要设置副本因子的文件都已经缓存到waitList列表中，下面看调用该方法对waitList列表中的文件执行的操作：
/*** 等待在waitList列表中的文件,所对应的每一个块的副本因子，都设置为指定的值rep*/void waitForReplication(List<Path> waitList, int rep) throws IOException {for(Path f : waitList) {System.out.print("Waiting for " + f + " ...");System.out.flush();boolean printWarning = false; // 如果文件f对应的块超过rep，是否给出警告信息（需要减少块副本数量，直到等于rep）FileStatus status = fs.getFileStatus(f); // 获取当前文件系统fs上文件f对应的FileStatus信息long len = status.getLen(); // 文件f的长度for(boolean done = false; !done; ) {BlockLocation[] locations = fs.getFileBlockLocations(status, 0, len); // 在当前fs上获取文件f对应的全部块的位置信息对象（一个数组）int i = 0;for(; i < locations.length && locations.getHosts().length == rep; i++){// 遍历文件f的每个块if (!printWarning && locations.getHosts().length > rep) { // 如果文件f的某个块的位置信息locations中，主机列表长度（其实就是副本因子的值）大于待设置的副本因子repSystem.out.println("/nWARNING: the waiting time may be long for " + "DECREASING the number of replication."); // 打印警告信息，需要适当删除该块副本，以满足副本因子要求printWarning = true; // 对于同一个文件f，只打印一次警告信息（如果满足f中的条件时）}} // fordone = i == locations.length; // 对文件f对应的块都检查过以后，设置检查完成标志doneif (!done) { // 没有经过上述检查（文件f对应的块副本小于0的情况下）System.out.print(".");System.out.flush();try {Thread.sleep(10000);} catch (InterruptedException e) {}}}System.out.println(" done");}}
这里，有必要了解一下org.apache.hadoop.fs.BlockLocation的含义，可以看BlockLocation类定义的属性，如下所示：
private String[] hosts; //hostnames of datanodesprivate String[] names; //hostname:portNumber of datanodesprivate String[] topologyPaths; // full path name in network topologyprivate long offset;//offset of the of the block in the fileprivate long length;
可见，一个BlockLocation包含了一个文件的一个块的详细信息，包括这个块对应的全部副本（包含它本身） ，比如上述定义的有：所在主机、所在主机及其端口号、在网络拓扑结构中的全路径名称、块在文件中的偏移位置、块长度。显然，这些块副本长度和在文件中的偏移位置都是相同的，可以共享（分别对应length和offset属性），其他三个属性的信息就不相同了（可能存在某两个相同的情况）。
Hadoop文件系统中，一个文件对应多个块（Block），每个块默认大小设置为64M。那么，对于由多个块组成的文件来说，如果想要获取到该文件的全部块及其块副本的信息，就需要通过文件系统中文件的统计信息FileStatus来获取到一个BlockLocation[]，该数组中对应的全部快就能够构成完整的该文件。
下面通过形式化语言来表达一下上面的含义：
假设一个文件F由n个块组成，则分别为：
B(1)，B(2)，……，B(n)
假设默认块的大小为BS，那么B(1)~B(n-1)一定是大小相同的块，大小都等于BS，而B(i)<=BS，这是显而易见的。
文件F的每个块B(i)都被存储在指定主机的文件系统中，假设存储到了主机H(i)上。为了快速计算，需要快速定位到文件F的Bi块上，也就是需要进行流式读取获取到，那么F的块B(i)需要有一个记录其详细信息的结构，也就是Hadoop定义的BlockLocation。假设Bi对应的描述信息对象为BL(i)，那么BL(i)就包含了与块B(i)相关的全部块副本的信息，当然每个块副本同样包含与BL(i)，相同的描述信息的属性，只是属性值不同而已。
假设文件F对应的块B(i)一共具有m个副本：
BR1(i)，BR2(i)，……，BRm(i)
这些块副本分别存储在对应如下的主机上：
H1(i)，H2(i)，……，Hm(i)
这些块副本分别对应指定主机的端口号分别如下：
H1(i):P1(i)，H2(i):P2(i)，……，Hm(i):Pm(i)
这些块副本对应的拓扑网络中的完整路径分别为：
U1(i)，U2(i)，……，Um(i)
假设块Bi的长度为LENGTH(i)，偏移位置为OFFSET(i)，那么，通过该文件的FileStatus获取的BlockLocation的内容，形式化的可以描述为：
new BlockLocation[]{newString{H1(i), H2(i), ……, Hm(i)},newString{H1(i):P1(i), H2(i):P2(i), ……, Hm(i):Pm(i)},newString{U1(i), U2(i), ……, Um(i)},LENGTH(i),OFFSET(i)}
关于获取到一个文件（对应的FileStatus）的BlockLocation[]，可以看到FileSystem类中getFileBlockLocations方法的实现，如下所示：
public BlockLocation[] getFileBlockLocations(FileStatus file, long start, long len) throws IOException { // 根据文件F对应的FileStatus，及其位置start和长度信息len就能获取到if (file == null) {return null;}if ( (start<0) || (len < 0) ) {throw new IllegalArgumentException("Invalid start or len parameter");}if (file.getLen() < start) {return new BlockLocation;}String[] name = { "localhost:50010" };String[] host = { "localhost" };return new BlockLocation[] { new BlockLocation(name, host, 0, file.getLen()) };}
上面这个方法只能获取到本机上的一个块。如果在Hadoop分布式文件系统中，这个方法就需要被重写了，使得通过客户端能够获取到指定文件的块，在不同主机上分布的块副本。
版权声明：本文为博主原创文章，未经博主允许不得转载。
页:
[1]