文件上传源码分析

先上文件上传的方法调用过程时序图
HDFS文件上传下载源码分析 - 图1

其主要执行过程：

1、FileSystem初始化，Client拿到NameNodeRpcServer代理对象，建立与NameNode的RPC通信

2、调用FileSystem的create()方法，由于实现类为DistributedFileSystem,所以是调用该类中的create()方法

3、DistributedFileSystem持有DFSClient的引用，继续调用DFSClient中的create()方法

4、DFSOutputStream提供的静态newStreamForCreate()方法中调用NameNodeRpcServer服务端的create()方法并创建DFSOutputStream输出流对象返回

5、通过hadoop提供的IOUtil工具类将输出流输出到本地

下面我们来看下源码：

首先初始化文件系统，建立与服务端的RPC通信：

调用FileSystem的create()方法，由于FileSystem是一个抽象类，这里实际上是调用的该类的子类create()方法

  1. public abstract FSDataOutputStream create(
  2.   Path var1, 
  3.   FsPermission var2, 
  4.   boolean var3, 
  5.   int var4, 
  6.   short var5, 
  7.   long var6, 
  8.   Progressable var8) throws IOException;

前面我们已经说过FileSystem.get()返回的是DistributedFileSystem对象，所以这里我们直接进入DistributedFileSystem：

  1. FileSystem fs=FileSystem.get(new URI("hdfs://hadoop01:9000"), conf, "hadoop");
  2. // 获取hdfs的操作对象   获取的是hdfs的操作句柄对象  FileSystem对象
  3. FileSystem fileSystem = FileSystem.get(conf);
  4. // DFS[DFSClient[clientName=DFSClient_NONMAPREDUCE_-22357267_1, ugi=hadoop (auth:SIMPLE)]] 
  5. System.out.println(fileSystem);
  7. //返回的是DistributedFileSystem对象
  8. System.out.println(fs instanceof DistributedFileSystem);
  1. @Override
  2.   public FSDataOutputStream create(final Path f, final FsPermission permission,
  3.       final EnumSet<CreateFlag> cflags, final int bufferSize,
  4.       final short replication, final long blockSize,
  5.       final Progressable progress, final ChecksumOpt checksumOpt)
  6.       throws IOException {
  7.     statistics.incrementWriteOps(1);
  8.     storageStatistics.incrementOpCounter(OpType.CREATE);
  9.     Path absF = fixRelativePart(f);
  10.     return new FileSystemLinkResolver<FSDataOutputStream>() {
  11.       @Override
  12.       public FSDataOutputStream doCall(final Path p) throws IOException {
  13.         final DFSOutputStream dfsos = dfs.create(getPathName(p), permission,
  14.             cflags, replication, blockSize, progress, bufferSize,
  15.             checksumOpt);
  16.         //dfs为DistributedFileSystem所持有的DFSClient对象,调用的是DFSClient.create方法
  17.         return dfs.createWrappedOutputStream(dfsos, statistics);
  18.       }
  19.       @Override
  20.       public FSDataOutputStream next(final FileSystem fs, final Path p)
  21.           throws IOException {
  22.         return fs.create(p, permission, cflags, bufferSize,
  23.             replication, blockSize, progress, checksumOpt);
  24.       }
  25.     }.resolve(this, absF);
  26.   }

DFSClient的create()返回一个DFSOutputStream对象：

  1. public DFSOutputStream create(String src, FsPermission permission,
  2.       EnumSet<CreateFlag> flag, boolean createParent, short replication,
  3.       long blockSize, Progressable progress, int buffersize,
  4.       ChecksumOpt checksumOpt, InetSocketAddress[] favoredNodes,
  5.       String ecPolicyName) throws IOException {
  6.     checkOpen();
  7.     final FsPermission masked = applyUMask(permission);
  8.     LOG.debug("{}: masked={}", src, masked);
  9.       //调用DFSOutputStream的静态方法newStreamForCreate，返回输出流
  10.     final DFSOutputStream result = DFSOutputStream.newStreamForCreate(this,
  11.         src, masked, flag, createParent, replication, blockSize, progress,
  12.         dfsClientConf.createChecksum(checksumOpt),
  13.         getFavoredNodesStr(favoredNodes), ecPolicyName);
  14.     beginFileLease(result.getFileId(), result);
  15.     return result;
  16.   }

我们继续看下newStreamForCreate()中的业务逻辑：

  1. tatic DFSOutputStream newStreamForCreate(DFSClient dfsClient, String src,
  2.       FsPermission masked, EnumSet<CreateFlag> flag, boolean createParent,
  3.       short replication, long blockSize, Progressable progress,
  4.       DataChecksum checksum, String[] favoredNodes, String ecPolicyName)
  5.       throws IOException {
  6.     try (TraceScope ignored =
  7.              dfsClient.newPathTraceScope("newStreamForCreate", src)) {
  8.       HdfsFileStatus stat = null;
  10.       // Retry the create if we get a RetryStartFileException up to a maximum
  11.       // number of times
  12.       boolean shouldRetry = true;
  13.       int retryCount = CREATE_RETRY_COUNT;
  14.       while (shouldRetry) {
  15.         shouldRetry = false;
  16.         try {
  17.           //通过dfsClient的 namenode 代理对象调用NamenodeRpcServer实现的create方法
  18.           stat = dfsClient.namenode.create(src, masked, dfsClient.clientName,
  19.               new EnumSetWritable<>(flag), createParent, replication,
  20.               blockSize, SUPPORTED_CRYPTO_VERSIONS, ecPolicyName);
  21.           break;
  22.         } catch (RemoteException re) {
  23.           IOException e = re.unwrapRemoteException(
  24.               AccessControlException.class,
  25.               DSQuotaExceededException.class,
  26.               QuotaByStorageTypeExceededException.class,
  27.               FileAlreadyExistsException.class,
  28.               FileNotFoundException.class,
  29.               ParentNotDirectoryException.class,
  30.               NSQuotaExceededException.class,
  31.               RetryStartFileException.class,
  32.               SafeModeException.class,
  33.               UnresolvedPathException.class,
  34.               SnapshotAccessControlException.class,
  35.               UnknownCryptoProtocolVersionException.class);
  36.           if (e instanceof RetryStartFileException) {
  37.             if (retryCount > 0) {
  38.               shouldRetry = true;
  39.               retryCount--;
  40.             } else {
  41.               throw new IOException("Too many retries because of encryption" +
  42.                   " zone operations", e);
  43.             }
  44.           } else {
  45.             throw e;
  46.           }
  47.         }
  48.       }
  49.       Preconditions.checkNotNull(stat, "HdfsFileStatus should not be null!");
  50.       //new 出输出流对象
  51.       final DFSOutputStream out;
  52.       if(stat.getErasureCodingPolicy() != null) {
  53.         out = new DFSStripedOutputStream(dfsClient, src, stat,
  54.             flag, progress, checksum, favoredNodes);
  55.       } else {
  56.         out = new DFSOutputStream(dfsClient, src, stat,
  57.             flag, progress, checksum, favoredNodes, true);
  58.       }
  59.       //调用内部类DataStreamer的start()方法，DataStreamer继承 Thread，所以这是一个线程，从 Namenode 中申请新的 block 信息
  60.       //同时之前提到的 pipline(流水线)作业也是在这里实现的。
  61.       out.start();
  62.       return out;
  63.     }
  64.   }

到此，Client拿到了服务端的输出流对象，那么后面就容易了，都是一些简答的文件输出，输入流的操作（hadoop提供的IOUitl）。

文件下载源码分析

文件上传的大致流程与文件下载类似，与上传一样，我们先上程序方法调用时序图：

HDFS文件上传下载源码分析 - 图2

主要执行过程：

  1. FileSystem初始化，Client拿到NameNodeRpcServer代理对象，建立与NameNode的RPC通信（与前面一样）

  2. 调用FileSystem的open()方法，由于实现类为DistributedFileSystem,所有是调用该类中的open()方法

  3. DistributedFileSystem持有DFSClient的引用，继续调用DFSClient中的open()方法

  4. 实例化DFSInputStream输入流

  5. 调用openinfo()方法

  6. 调用fetchLocatedBlocksAndGetLastBlockLength()方法，抓取block信息并获取最后block长度

  7. 调用DFSClient中的getLocatedBlocks()方法，获取block信息

  8. 在callGetBlockLocations()方法中通过NameNode代理对象调用NameNodeRpcServer的getBlockLocations()方法

  9. 将block信息写入输出流

  10. 交给IOUtil,下载文件到本地

接下来，我们开始看源码：

首先任然是FileSystem的初始化，前面有，这里就不贴出来了，我们直接从DistributedFileSystem的open()开始看。

  1. @Override
  2.   public FSDataInputStream open(PathHandle fd, int bufferSize)
  3.       throws IOException {
  4.     statistics.incrementReadOps(1);
  5.     storageStatistics.incrementOpCounter(OpType.OPEN);
  6.     if (!(fd instanceof HdfsPathHandle)) {
  7.       fd = new HdfsPathHandle(fd.bytes());
  8.     }
  9.     HdfsPathHandle id = (HdfsPathHandle) fd;
  10.     final DFSInputStream dfsis = dfs.open(id, bufferSize, verifyChecksum);
  11.     //dfs 为 DFSClient对象，调用 open()防护输入流
  12.     return dfs.createWrappedInputStream(dfsis);
  13.   }

DFSClient中并没有直接使用NameNode的代理对象，而是传给了DFSInputStream：

  1. public DFSInputStream open(String src, int buffersize, boolean verifyChecksum)
  2.   throws IOException {
  3.   checkOpen();
  4.   //    Get block info from namenode
  5.   try (TraceScope ignored = newPathTraceScope("newDFSInputStream", src)) {
  6.     LocatedBlocks locatedBlocks = getLocatedBlocks(src, 0);
  8.     return openInternal(locatedBlocks, src, verifyChecksum);
  9.   }
  10. }
  12. private DFSInputStream openInternal(LocatedBlocks locatedBlocks, String src,
  13.                                     boolean verifyChecksum) throws IOException {
  14.   if (locatedBlocks != null) {
  15.     ErasureCodingPolicy ecPolicy = locatedBlocks.getErasureCodingPolicy();
  16.     if (ecPolicy != null) {
  17.       return new DFSStripedInputStream(this, src, verifyChecksum, ecPolicy,
  18.                                        locatedBlocks);
  19.     }
  20.     //DFSClient中并没有直接使用NameNode的代理对象调用服务端方法，直接 new 输入流并把当前对象作为参数传入
  21.     return new DFSInputStream(this, src, verifyChecksum, locatedBlocks);
  22.   } else {
  23.     throw new IOException("Cannot open filename " + src);
  24.   }
  25. }

那么在DFSInputStream必须持有DFSClient的引用：

  1. //DFSInputStream.java 构造
  2. DFSInputStream(DFSClient dfsClient, String src, boolean verifyChecksum,
  3.       LocatedBlocks locatedBlocks) throws IOException {
  4.     this.dfsClient = dfsClient; //DFSClient的引用
  5.     this.verifyChecksum = verifyChecksum;
  6.     this.src = src;
  7.     synchronized (infoLock) {
  8.       this.cachingStrategy = dfsClient.getDefaultReadCachingStrategy();
  9.     }
  10.     this.locatedBlocks = locatedBlocks;
  11.     openInfo(false);
  12.   }

openInfo()用来抓取block信息：

  1. //DFSInputStream.java
  2. void openInfo(boolean refreshLocatedBlocks) throws IOException {
  3.     final DfsClientConf conf = dfsClient.getConf();
  4.     synchronized(infoLock) {
  5.       //获取 block的信息
  6.       lastBlockBeingWrittenLength =
  7.           fetchLocatedBlocksAndGetLastBlockLength(refreshLocatedBlocks);
  8.       //获取配置信息，尝试抓取的次数，默认是 3
  9.       int retriesForLastBlockLength = conf.getRetryTimesForGetLastBlockLength();
  10.       while (retriesForLastBlockLength > 0) {
  11.         // Getting last block length as -1 is a special case. When cluster
  12.         // restarts, DNs may not report immediately. At this time partial block
  13.         // locations will not be available with NN for getting the length. Lets
  14.         // retry for 3 times to get the length.
  15.         if (lastBlockBeingWrittenLength == -1) {
  16.           DFSClient.LOG.warn("Last block locations not available. "
  17.               + "Datanodes might not have reported blocks completely."
  18.               + " Will retry for " + retriesForLastBlockLength + " times");
  19.           waitFor(conf.getRetryIntervalForGetLastBlockLength());
  20.           lastBlockBeingWrittenLength =
  21.               fetchLocatedBlocksAndGetLastBlockLength(true);
  22.         } else {
  23.           break;
  24.         }
  25.         retriesForLastBlockLength--;
  26.       }
  27.       if (lastBlockBeingWrittenLength == -1
  28.           && retriesForLastBlockLength == 0) {
  29.         throw new IOException("Could not obtain the last block locations.");
  30.       }
  31.     }
  32.   }

获取block信息：

  1.   private long fetchLocatedBlocksAndGetLastBlockLength(boolean refresh)
  2.       throws IOException {
  3.     LocatedBlocks newInfo = locatedBlocks;
  4.     //回到 DFSClient 中获取当前 block的信息
  5.     if (locatedBlocks == null || refresh) {
  6.       newInfo = dfsClient.getLocatedBlocks(src, 0);
  7.     }
  8.     DFSClient.LOG.debug("newInfo = {}", newInfo);
  9.     if (newInfo == null) {
  10.       throw new IOException("Cannot open filename " + src);
  11.     }
  13.     if (locatedBlocks != null) {
  14.       Iterator<LocatedBlock> oldIter = locatedBlocks.getLocatedBlocks().iterator();
  15.       Iterator<LocatedBlock> newIter = newInfo.getLocatedBlocks().iterator();
  16.       while (oldIter.hasNext() && newIter.hasNext()) {
  17.         if (!oldIter.next().getBlock().equals(newIter.next().getBlock())) {
  18.           throw new IOException("Blocklist for " + src + " has changed!");
  19.         }
  20.       }
  21.     }
  22.     locatedBlocks = newInfo;
  23.     long lastBlockBeingWrittenLength = 0;
  24.     if (!locatedBlocks.isLastBlockComplete()) {
  25.       final LocatedBlock last = locatedBlocks.getLastLocatedBlock();
  26.       if (last != null) {
  27.         if (last.getLocations().length == 0) {
  28.           if (last.getBlockSize() == 0) {
  29.             // if the length is zero, then no data has been written to
  30.             // datanode. So no need to wait for the locations.
  31.             return 0;
  32.           }
  33.           return -1;
  34.         }
  35.         final long len = readBlockLength(last);
  36.         last.getBlock().setNumBytes(len);
  37.         lastBlockBeingWrittenLength = len;
  38.       }
  39.     }
  41.     fileEncryptionInfo = locatedBlocks.getFileEncryptionInfo();
  42.         // 返回 block 开始写的位置
  43.     return lastBlockBeingWrittenLength;
  44.   }

回到DFSClient中：

  1. @VisibleForTesting
  2. public LocatedBlocks getLocatedBlocks(String src, long start, long length)
  3.   throws IOException {
  4.   try (TraceScope ignored = newPathTraceScope("getBlockLocations", src)) {
  5.     //这里的namenode作为参数传递到callGetBlockLocations（）中
  6.     return callGetBlockLocations(namenode, src, start, length);
  7.   }
  8. }

调用服务端方法，返回block信息：

  1. static LocatedBlocks callGetBlockLocations(ClientProtocol namenode,
  2.                                            String src, long start, long length)
  3.   throws IOException {
  4.   try {
  5.     // 返回 block信息
  6.     return namenode.getBlockLocations(src, start, length);
  7.   } catch(RemoteException re) {
  8.     throw re.unwrapRemoteException(AccessControlException.class,
  9.                                    FileNotFoundException.class,
  10.                                    UnresolvedPathException.class);
  11.   }
  12. }

最终将文件block相关信息写入输入流，通过工具类IOUtil输出到本地文件。

文章来自公众号：JS ONE 技术联盟