Hadoop之HDFS原理及文件上傳下載源碼分析（下）

　　上篇Hadoop之HDFS原理及文件上傳下載源碼分析（上）樓主主要介紹了hdfs原理及FileSystem的初始化源碼解析， Client如何與NameNode建立RPC通信。本篇將繼續介紹hdfs文件上傳、下載源解析。

文件上傳

　　先上文件上傳的方法調用過程時序圖：

　　

　　

 　　其主要執行過程：

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通信

　　

HDFSDemo.java
OutputStream os = fs.create(new Path("/test.log"));

 

　　調用FileSystem的create()方法，由於FileSystem是一個抽象類，這里實際上是調用的該類的子類create()方法

　　

 //FileSystem.java
public abstract FSDataOutputStream create(Path f,
      FsPermission permission,
      boolean overwrite,
      int bufferSize,
      short replication,
      long blockSize,
      Progressable progress) throws IOException;

　　　前面我們已經說過FileSystem.get()返回的是DistributedFileSystem對象，所以這里我們直接進入DistributedFileSystem：

　

　　

  //DistributedFileSystem.java
@Override
  public FSDataOutputStream create(final Path f, final FsPermission permission,
    final EnumSet<CreateFlag> cflags, final int bufferSize,
    final short replication, final long blockSize, final Progressable progress,
    final ChecksumOpt checksumOpt) throws IOException {
    statistics.incrementWriteOps(1);
    Path absF = fixRelativePart(f);
    return new FileSystemLinkResolver<FSDataOutputStream>() {
      @Override
      public FSDataOutputStream doCall(final Path p)
          throws IOException, UnresolvedLinkException {
        final DFSOutputStream dfsos = dfs.create(getPathName(p), permission,
                cflags, replication, blockSize, progress, bufferSize,
                checksumOpt);
        //dfs為DistributedFileSystem所持有的DFSClient對象，這里調用DFSClient中的create()方法
        return dfs.createWrappedOutputStream(dfsos, statistics);
      }
      @Override
      public FSDataOutputStream next(final FileSystem fs, final Path p)
          throws IOException {
        return fs.create(p, permission, cflags, bufferSize,
            replication, blockSize, progress, checksumOpt);
      }
    }.resolve(this, absF);
  }

　　DFSClient的create()返回一個DFSOutputStream對象：

　　

 //DFSClient.java
public DFSOutputStream create(String src, 
                             FsPermission permission,
                             EnumSet<CreateFlag> flag, 
                             boolean createParent,
                             short replication,
                             long blockSize,
                             Progressable progress,
                             int buffersize,
                             ChecksumOpt checksumOpt,
                             InetSocketAddress[] favoredNodes) throws IOException {
    checkOpen();
    if (permission == null) {
      permission = FsPermission.getFileDefault();
    }
    FsPermission masked = permission.applyUMask(dfsClientConf.uMask);
    if(LOG.isDebugEnabled()) {
      LOG.debug(src + ": masked=" + masked);
    }
    //調用DFSOutputStream的靜態方法newStreamForCreate，返回輸出流
    final DFSOutputStream result = DFSOutputStream.newStreamForCreate(this,
        src, masked, flag, createParent, replication, blockSize, progress,
        buffersize, dfsClientConf.createChecksum(checksumOpt),
        getFavoredNodesStr(favoredNodes));
    beginFileLease(result.getFileId(), result);
    return result;
  }

　　我們繼續看下newStreamForCreate()中的業務邏輯：

　　

//DFSOutputStream.java
 static DFSOutputStream newStreamForCreate(DFSClient dfsClient, String src,
      FsPermission masked, EnumSet<CreateFlag> flag, boolean createParent,
      short replication, long blockSize, Progressable progress, int buffersize,
      DataChecksum checksum, String[] favoredNodes) throws IOException {
    TraceScope scope =
        dfsClient.getPathTraceScope("newStreamForCreate", src);
    try {
      HdfsFileStatus stat = null;
      boolean shouldRetry = true;
      int retryCount = CREATE_RETRY_COUNT;
      while (shouldRetry) {
        shouldRetry = false;
        try {
          //這里通過dfsClient的NameNode代理對象調用NameNodeRpcServer中實現的create()方法
          stat = dfsClient.namenode.create(src, masked, dfsClient.clientName,
              new EnumSetWritable<CreateFlag>(flag), createParent, replication,
              blockSize, SUPPORTED_CRYPTO_VERSIONS);
          break;
        } catch (RemoteException re) {
          IOException e = re.unwrapRemoteException(
              AccessControlException.class,
              DSQuotaExceededException.class,
              FileAlreadyExistsException.class,
              FileNotFoundException.class,
              ParentNotDirectoryException.class,
              NSQuotaExceededException.class,
              RetryStartFileException.class,
              SafeModeException.class,
              UnresolvedPathException.class,
              SnapshotAccessControlException.class,
              UnknownCryptoProtocolVersionException.class);
          if (e instanceof RetryStartFileException) {
            if (retryCount > 0) {
              shouldRetry = true;
              retryCount--;
            } else {
              throw new IOException("Too many retries because of encryption" +
                  " zone operations", e);
            }
          } else {
            throw e;
          }
        }
      }
      Preconditions.checkNotNull(stat, "HdfsFileStatus should not be null!");
     //new輸出流對象
      final DFSOutputStream out = new DFSOutputStream(dfsClient, src, stat,
          flag, progress, checksum, favoredNodes);
      out.start();//調用內部類DataStreamer的start()方法，DataStreamer繼承Thread，所以說這是一個線程，從NameNode中申請新的block信息；
　　　　　　　　　　　　　　　　同時前面我們介紹hdfs原理的時候提到的流水線作業（Pipeline）也是在這里實現，有興趣的同學可以去研究下，這里就不帶大家看了
      return out;
    } finally {
      scope.close();
    }
  }

　　　　

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

文件下載

　　文件上傳的大致流程與文件下載類似，與上傳一樣，我們先上程序方法調用時序圖：

　　

　　主要執行過程：　　

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()開始看。

　　

//DistributedFifeSystem.java
@Override
  public FSDataInputStream open(Path f, final int bufferSize)
      throws IOException {
    statistics.incrementReadOps(1);
    Path absF = fixRelativePart(f);
    return new FileSystemLinkResolver<FSDataInputStream>() {
      @Override
      public FSDataInputStream doCall(final Path p)
          throws IOException, UnresolvedLinkException {
        final DFSInputStream dfsis =
          dfs.open(getPathName(p), bufferSize, verifyChecksum);
        //dfs為DFSClient對象，調用open()返回輸入流
        return dfs.createWrappedInputStream(dfsis);
      }
      @Override
      public FSDataInputStream next(final FileSystem fs, final Path p)
          throws IOException {
        return fs.open(p, bufferSize);
      }
    }.resolve(this, absF);
  }

　　DFSClient中並沒有直接使用NameNode的代理對象，而是傳給了DFSInputStream：

　　

//DFSClient.java
public DFSInputStream open(String src, int buffersize, boolean verifyChecksum)
      throws IOException, UnresolvedLinkException {
    checkOpen();   
    TraceScope scope = getPathTraceScope("newDFSInputStream", src);
    try {
      //這里並沒有直接通過NameNode的代理對象調用服務端的方法，直接new輸入流並把當前對象作為參數傳入
      return new DFSInputStream(this, src, verifyChecksum);
    } finally {
      scope.close();
    }
  }

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

　　

//DFSInputStream.java 構造
DFSInputStream(DFSClient dfsClient, String src, boolean verifyChecksum
                 ) throws IOException, UnresolvedLinkException {
    this.dfsClient = dfsClient;//只有DFSClient的引用
    this.verifyChecksum = verifyChecksum;
    this.src = src;
    synchronized (infoLock) {
      this.cachingStrategy = dfsClient.getDefaultReadCachingStrategy();
    }
    openInfo();//調openInfo()
  }

　　openInfo()用來抓取block信息：

void openInfo() throws IOException, UnresolvedLinkException {
    synchronized(infoLock) {
      lastBlockBeingWrittenLength = fetchLocatedBlocksAndGetLastBlockLength();//抓取block信息
      int retriesForLastBlockLength = dfsClient.getConf().retryTimesForGetLastBlockLength;//獲取配置信息，嘗試抓取的次數，樓主記得在2.6以前這里寫的3；當然，現在的默認值也為3
      while (retriesForLastBlockLength > 0) {
        if (lastBlockBeingWrittenLength == -1) {
          DFSClient.LOG.warn("Last block locations not available. "
              + "Datanodes might not have reported blocks completely."
              + " Will retry for " + retriesForLastBlockLength + " times");
          waitFor(dfsClient.getConf().retryIntervalForGetLastBlockLength);
          lastBlockBeingWrittenLength = fetchLocatedBlocksAndGetLastBlockLength();
        } else {
          break;
        }
        retriesForLastBlockLength--;
      }
      if (retriesForLastBlockLength == 0) {
        throw new IOException("Could not obtain the last block locations.");
      }
    }
  }

 

　　獲取block信息：

//DFSInputStream.java
private long fetchLocatedBlocksAndGetLastBlockLength() throws IOException {
    final LocatedBlocks newInfo = dfsClient.getLocatedBlocks(src, 0);
    //回到DFSClient中來獲取當前block信息
    if (DFSClient.LOG.isDebugEnabled()) {
      DFSClient.LOG.debug("newInfo = " + newInfo);
    }
    if (newInfo == null) {
      throw new IOException("Cannot open filename " + src);
    }

    if (locatedBlocks != null) {
      Iterator<LocatedBlock> oldIter = locatedBlocks.getLocatedBlocks().iterator();
      Iterator<LocatedBlock> newIter = newInfo.getLocatedBlocks().iterator();
      while (oldIter.hasNext() && newIter.hasNext()) {
        if (! oldIter.next().getBlock().equals(newIter.next().getBlock())) {
          throw new IOException("Blocklist for " + src + " has changed!");
        }
      }
    }
    locatedBlocks = newInfo;
    long lastBlockBeingWrittenLength = 0;
    if (!locatedBlocks.isLastBlockComplete()) {
      final LocatedBlock last = locatedBlocks.getLastLocatedBlock();
      if (last != null) {
        if (last.getLocations().length == 0) {
          if (last.getBlockSize() == 0) {         
            return 0;
          }
          return -1;
        }
        final long len = readBlockLength(last);
        last.getBlock().setNumBytes(len);
        lastBlockBeingWrittenLength = len; 
      }
    }

    fileEncryptionInfo = locatedBlocks.getFileEncryptionInfo();
    //返回block開始寫的位置
    return lastBlockBeingWrittenLength;
  }

　　回到DFSClient中：

　　

DFSClient.java
@VisibleForTesting
  public LocatedBlocks getLocatedBlocks(String src, long start, long length)
      throws IOException {
    TraceScope scope = getPathTraceScope("getBlockLocations", src);
    try {
      //這里NameNode作為參數傳遞到callGetBlockLocations()中
      return callGetBlockLocations(namenode, src, start, length);
    } finally {
      scope.close();
    }
  }

　　調用服務端方法，返回block信息：

//DFSClient.java
static LocatedBlocks callGetBlockLocations(ClientProtocol namenode,
      String src, long start, long length) 
      throws IOException {
    try {
     //看到這里，不用做過多的解釋了吧？
      return namenode.getBlockLocations(src, start, length);
    } catch(RemoteException re) {
      throw re.unwrapRemoteException(AccessControlException.class,
                                     FileNotFoundException.class,
                                     UnresolvedPathException.class);
    }
  }

　　最終將文件block相關信息寫入輸入流，通過工具類IOUtil輸出到本地文件。

　　那關於hadoop之hdfs原理及文件上傳下載源碼解析就寫到這里，下系列的文章，樓主會寫一些關於mapreduce或者hive相關的文章分享給大家。

　　示例代碼地址：https://github.com/LJunChina/hadoop