• HDFS文件上传之create创建过程-尚硅谷大数据培训

    HDFS的写数据流程，如下图所示：

    HDFS上传源码解析如下图所示：

    0）在pom.xml中增加如下依赖

    1. <dependencies>
    2.        <dependency>
    3.               <groupId>org.apache.hadoopgroupId>
    4.               <artifactId>hadoop-clientartifactId>
    5.               <version>3.1.3version>
    6.        dependency>
    7.  
    8.        <dependency>
    9.               <groupId>org.apache.hadoopgroupId>
    10.               <artifactId>hadoop-hdfsartifactId>
    11.               <version>3.1.3version>
    12.        dependency>
    13.  
    14.        <dependency>
    15.               <groupId>org.apache.hadoopgroupId>
    16.               <artifactId>hadoop-hdfs-clientartifactId>
    17.               <version>3.1.3version>
    18.               <scope>providedscope>
    19.        dependency>
    20.  
    21.        <dependency>
    22.               <groupId>junitgroupId>
    23.               <artifactId>junitartifactId>
    24.               <version>4.12version>
    25.        dependency>
    26.        <dependency>
    27.               <groupId>org.slf4jgroupId>
    28.               <artifactId>slf4j-log4j12artifactId>
    29.               <version>1.7.30version>
    30.        dependency>
    31. dependencies>

    一、DN向NN发起创建请求

    用户自己写的代码

    1. @Test
    2. public void testPut2() throws IOException {
    3.        FSDataOutputStream fos = fs.create(newPath("/input"));
    4.  
    5.        fos.write("helloworld".getBytes());
    6. }

    FileSystem.java

    1. public FSDataOutputStream create(Path f) throws IOException {
    2.        return create(f, true);
    3. }
    4.  
    5. public FSDataOutputStream create(Path f, booleanoverwrite)
    6.   throws IOException {
    7.  
    8.        return create(f, overwrite,
    9.                        getConf().getInt(IO_FILE_BUFFER_SIZE_KEY,
    10.                               IO_FILE_BUFFER_SIZE_DEFAULT),
    11.                        getDefaultReplication(f),
    12.                        getDefaultBlockSize(f));
    13. }
    14.  
    15. public FSDataOutputStream create(Path f,
    16.        boolean overwrite,
    17.        int bufferSize,
    18.        short replication,
    19.        long blockSize) throws IOException {
    20.       
    21.        return create(f, overwrite, bufferSize, replication, blockSize,null);
    22. }
    23.  
    24. public FSDataOutputStream create(Path f,
    25.        boolean overwrite,
    26.        int bufferSize,
    27.        short replication,
    28.        long blockSize,
    29.        Progressable progress
    30.        ) throws IOException {
    31.                                                                      
    32.        return this.create(f, FsCreateModes.applyUMask(
    33.        FsPermission.getFileDefault(),FsPermission.getUMask(getConf())),
    34.        overwrite, bufferSize, replication,blockSize, progress);
    35. }
    36.  
    37. public abstract FSDataOutputStream create(Path f,
    38.        FsPermission permission,
    39.        boolean overwrite,
    40.        int bufferSize,
    41.        short replication,
    42.        long blockSize,
    43.        Progressable progress) throwsIOException;

    选中create，点击ctrl+h，找到实现类
    DistributedFileSystem.java，查找create方法。

    DistributedFileSystem.java

    1. @Override
    2. public FSDataOutputStream create(Path f,FsPermission permission,
    3.   boolean overwrite, int bufferSize, shortreplication, long blockSize,
    4.   Progressable progress) throws IOException {
    5.  
    6.        return this.create(f, permission,
    7.        overwrite ? EnumSet.of(CreateFlag.CREATE,CreateFlag.OVERWRITE)
    8.               : EnumSet.of(CreateFlag.CREATE),bufferSize, replication,
    9.        blockSize, progress, null);
    10. }
    11.  
    12. @Override
    13. public FSDataOutputStreamcreate(final Path f, finalFsPermission permission,
    14.   final EnumSet cflags, finalint bufferSize,
    15.   final short replication, final longblockSize,
    16.   final Progressable progress, finalChecksumOpt checksumOpt)
    17.   throws IOException {
    18.  
    19.        statistics.incrementWriteOps(1);
    20.        storageStatistics.incrementOpCounter(OpType.CREATE);
    21.        Path absF = fixRelativePart(f);
    22.       
    23.        return newFileSystemLinkResolver() {
    24.  
    25.         @Override
    26.         public FSDataOutputStream doCall(final Path p) throws IOException {
    27.  
    28.               // 创建获取了一个输出流对象
    29.               final DFSOutputStream dfsos = dfs.create(getPathName(p), permission,
    30.                      cflags, replication,blockSize, progress, bufferSize,
    31.                      checksumOpt);
    32.               // 这里将上面创建的dfsos进行包装并返回
    33.               return dfs.createWrappedOutputStream(dfsos,statistics);
    34.          }
    35.  
    36.         @Override
    37.         public FSDataOutputStream next(final FileSystem fs, final Path p)
    38.                throws IOException {
    39.               return fs.create(p, permission,cflags, bufferSize,
    40.                      replication, blockSize,progress, checksumOpt);
    41.          }
    42.        }.resolve(this, absF);
    43. }

    点击create，进入DFSClient.java

    1. public DFSOutputStream create(String src,FsPermission permission,
    2.   EnumSet flag, short replication, long blockSize,
    3.   Progressable progress, int buffersize, ChecksumOpt checksumOpt)
    4.   throws IOException {
    5.  
    6.        return create(src, permission, flag, true,
    7.        replication, blockSize, progress,buffersize, checksumOpt, null);
    8. }
    9.  
    10. public DFSOutputStream create(String src,FsPermission permission,
    11.   EnumSet flag, boolean createParent, short replication,
    12.   long blockSize, Progressable progress, int buffersize,
    13.   ChecksumOpt checksumOpt, InetSocketAddress[]favoredNodes)
    14.   throws IOException {
    15.  
    16.        return create(src, permission, flag, createParent, replication,blockSize,
    17.        progress, buffersize, checksumOpt,favoredNodes, null);
    18. }
    19.  
    20. public DFSOutputStream create(String src,FsPermission permission,
    21.   EnumSet flag, boolean createParent, short replication,
    22.   long blockSize, Progressable progress, int buffersize,
    23.   ChecksumOpt checksumOpt, InetSocketAddress[]favoredNodes,
    24.   String ecPolicyName) throws IOException {
    25.  
    26.        checkOpen();
    27.       
    28.        final FsPermission masked =applyUMask(permission);
    29.        LOG.debug("{}: masked={}", src,masked);
    30.       
    31.        final DFSOutputStream result = DFSOutputStream.newStreamForCreate(this,
    32.               src, masked, flag, createParent,replication, blockSize, progress,
    33.               dfsClientConf.createChecksum(checksumOpt),
    34.               getFavoredNodesStr(favoredNodes),ecPolicyName);
    35.              
    36.        beginFileLease(result.getFileId(),result);
    37.       
    38.        return result;
    39. }

    点击newStreamForCreate，进入DFSOutputStream.java

    1. static DFSOutputStream newStreamForCreate(DFSClientdfsClient, String src,
    2.   FsPermission masked,EnumSet flag, boolean createParent,
    3.   short replication, long blockSize,Progressable progress,
    4.   DataChecksum checksum, String[] favoredNodes,String ecPolicyName)
    5.   throws IOException {
    6.  
    7.        try (TraceScope ignored =
    8.                       dfsClient.newPathTraceScope("newStreamForCreate",src)) {
    9.          HdfsFileStatusstat = null;
    10.  
    11.         // Retry the create if we get a RetryStartFileException up to a maximum
    12.         // number of times
    13.         boolean shouldRetry = true;
    14.         int retryCount = CREATE_RETRY_COUNT;
    15.  
    16.         while (shouldRetry) {
    17.               shouldRetry = false;
    18.               try {
    19.                // DN将创建请求发送给NN（RPC）
    20.                stat = dfsClient.namenode.create(src, masked,dfsClient.clientName,
    21.                        new EnumSetWritable<>(flag),createParent, replication,
    22.                        blockSize, SUPPORTED_CRYPTO_VERSIONS,ecPolicyName);
    23.                break;
    24.               } catch (RemoteException re) {
    25.                … ….
    26.               }
    27.          }
    28.         Preconditions.checkNotNull(stat, "HdfsFileStatus should not benull!");
    29.         final DFSOutputStream out;
    30.  
    31.         if(stat.getErasureCodingPolicy() != null) {
    32.               out = newDFSStripedOutputStream(dfsClient, src, stat,
    33.                      flag, progress, checksum,favoredNodes);
    34.          }else {
    35.               out = newDFSOutputStream(dfsClient, src, stat,
    36.                      flag, progress, checksum,favoredNodes, true);
    37.          }
    38.  
    39.          //开启线程run，DataStreamer extends Daemon extends Thread
    40.          out.start();
    41.  
    42.         return out;
    43.        }
    44. }

    二、NN处理DN的创建请求

    1）点击create

    ClientProtocol.java

    1. HdfsFileStatus create(String src, FsPermissionmasked,
    2.     String clientName,EnumSetWritable flag,
    3.     boolean createParent, shortreplication, long blockSize,
    4.     CryptoProtocolVersion[]supportedVersions, String ecPolicyName)
    5.     throws IOException;

    2）Ctrl + h查找create实现类，点击NameNodeRpcServer，在NameNodeRpcServer.java中搜索create

    NameNodeRpcServer.java

    1. public HdfsFileStatuscreate(String src, FsPermissionmasked,
    2.     String clientName,EnumSetWritable flag,
    3.     boolean createParent, short replication,long blockSize,
    4.     CryptoProtocolVersion[] supportedVersions,String ecPolicyName)
    5.     throws IOException {
    6.   // 检查NN启动
    7.   checkNNStartup();
    8.   ... ...
    9.  
    10.   HdfsFileStatus status = null;
    11.   try {
    12.     PermissionStatus perm = newPermissionStatus(getRemoteUser()
    13.         .getShortUserName(), null, masked);
    14.        // 重要
    15.     status = namesystem.startFile(src, perm, clientName,clientMachine,
    16.         flag.get(), createParent, replication,blockSize, supportedVersions,
    17.         ecPolicyName, cacheEntry != null);
    18.   } finally {
    19.     RetryCache.setState(cacheEntry, status !=null, status);
    20.   }
    21.  
    22.   metrics.incrFilesCreated();
    23.   metrics.incrCreateFileOps();
    24.   return status;
    25. }

    FSNamesystem.java

    1. HdfsFileStatus startFile(String src, PermissionStatuspermissions,
    2.     String holder, String clientMachine,EnumSet flag,
    3.     boolean createParent, short replication,long blockSize,
    4.     CryptoProtocolVersion[] supportedVersions,String ecPolicyName,
    5.     boolean logRetryCache) throws IOException {
    6.  
    7.   HdfsFileStatus status;
    8.   try {
    9.     status = startFileInt(src, permissions, holder, clientMachine, flag,
    10.         createParent, replication, blockSize,supportedVersions, ecPolicyName,
    11.         logRetryCache);
    12.   } catch (AccessControlException e) {
    13.     logAuditEvent(false, "create",src);
    14.     throw e;
    15.   }
    16.   logAuditEvent(true, "create", src,status);
    17.   return status;
    18. }
    19.  
    20. private HdfsFileStatus startFileInt(String src,
    21.     PermissionStatus permissions, Stringholder, String clientMachine,
    22.     EnumSet flag, booleancreateParent, short replication,
    23.     long blockSize, CryptoProtocolVersion[]supportedVersions,
    24.     String ecPolicyName, boolean logRetryCache)throws IOException {      
    25.        ... ...
    26.        stat = FSDirWriteFileOp.startFile(this, iip, permissions,holder,
    27.         clientMachine, flag, createParent,replication, blockSize, feInfo,
    28.         toRemoveBlocks, shouldReplicate,ecPolicyName, logRetryCache);
    29.        ... ...
    30. }
    31.  
    32. static HdfsFileStatus startFile(
    33.     ... ...)
    34.     throws IOException {
    35.       
    36.   ... ...
    37.   FSDirectory fsd = fsn.getFSDirectory();
    38.  
    39.   // 文件路径是否存在校验
    40.   if (iip.getLastINode() != null) {
    41.     if (overwrite) {
    42.       List toRemoveINodes = newChunkedArrayList<>();
    43.       List toRemoveUCFiles = newChunkedArrayList<>();
    44.       long ret = FSDirDeleteOp.delete(fsd, iip,toRemoveBlocks,
    45.                                      toRemoveINodes, toRemoveUCFiles, now());
    46.       if (ret >= 0) {
    47.         iip = INodesInPath.replace(iip,iip.length() - 1, null);
    48.        FSDirDeleteOp.incrDeletedFileCount(ret);
    49.        fsn.removeLeasesAndINodes(toRemoveUCFiles, toRemoveINodes, true);
    50.       }
    51.     } else {
    52.       // If lease soft limit time is expired,recover the lease
    53.      fsn.recoverLeaseInternal(FSNamesystem.RecoverLeaseOp.CREATE_FILE, iip,
    54.                                src, holder,clientMachine, false);
    55.       throw new FileAlreadyExistsException(src + " forclient " +
    56.          clientMachine + " already exists");
    57.     }
    58.   }
    59.   fsn.checkFsObjectLimit();
    60.   INodeFile newNode = null;
    61.   INodesInPath parent = FSDirMkdirOp.createAncestorDirectories(fsd,iip, permissions);
    62.   if (parent != null) {
    63.     // 添加文件元数据信息
    64.     iip = addFile(fsd, parent, iip.getLastLocalName(), permissions,
    65.         replication, blockSize, holder,clientMachine, shouldReplicate,
    66.         ecPolicyName);
    67.     newNode = iip != null ?iip.getLastINode().asFile() : null;
    68.   }
    69.   ... ...
    70.  setNewINodeStoragePolicy(fsd.getBlockManager(), iip, isLazyPersist);
    71.   fsd.getEditLog().logOpenFile(src, newNode,overwrite, logRetryEntry);
    72.   if (NameNode.stateChangeLog.isDebugEnabled()){
    73.     NameNode.stateChangeLog.debug("DIR*NameSystem.startFile: added " +
    74.         src + " inode " +newNode.getId() + " " + holder);
    75.   }
    76.   return FSDirStatAndListingOp.getFileInfo(fsd,iip, false, false);
    77. }
    78.  
    79. private static INodesInPath addFile(
    80.     FSDirectory fsd, INodesInPath existing,byte[] localName,
    81.     PermissionStatus permissions, shortreplication, long preferredBlockSize,
    82.     String clientName, String clientMachine,boolean shouldReplicate,
    83.     String ecPolicyName) throws IOException {
    84.  
    85.   Preconditions.checkNotNull(existing);
    86.   long modTime = now();
    87.   INodesInPath newiip;
    88.   fsd.writeLock();
    89.   try {
    90.     … …
    91.  
    92.     newiip = fsd.addINode(existing, newNode, permissions.getPermission());
    93.   } finally {
    94.     fsd.writeUnlock();
    95.   }
    96.   ... ...
    97.   return newiip;
    98. }
    99.  
    100. INodesInPath addINode(INodesInPath existing, INodechild,
    101.                       FsPermission modes)
    102.     throws QuotaExceededException,UnresolvedLinkException {
    103.   cacheName(child);
    104.   writeLock();
    105.   try {
    106.     // 将数据写入到INode的目录树中
    107.     return addLastINode(existing, child, modes, true);
    108.   } finally {
    109.     writeUnlock();
    110.   }
    111. }

    三、DataStreamer启动流程

    NN处理完DN请求后，再次回到DN端，启动对应的线程

    DFSOutputStream.java

    1. static DFSOutputStream newStreamForCreate(DFSClientdfsClient, String src,
    2.   FsPermission masked,EnumSet flag, boolean createParent,
    3.   short replication, long blockSize,Progressable progress,
    4.   DataChecksum checksum, String[] favoredNodes,String ecPolicyName)
    5.   throws IOException {
    6.        ... ...
    7.        // DN将创建请求发送给NN（RPC）
    8.        stat = dfsClient.namenode.create(src,masked, dfsClient.clientName,
    9.         new EnumSetWritable<>(flag), createParent, replication,
    10.         blockSize, SUPPORTED_CRYPTO_VERSIONS, ecPolicyName);
    11.        ... ...
    12.       
    13.        // 创建输出流
    14.        out = new DFSOutputStream(dfsClient, src, stat,
    15.             flag, progress, checksum,favoredNodes, true);
    16.        // 开启线程run，DataStreamer extends Daemon extends Thread
    17.        out.start();
    18.  
    19.        return out;
    20. }

    点击DFSOutputStream

    1. protected DFSOutputStream(DFSClient dfsClient, String src,
    2.     HdfsFileStatus stat,EnumSet<CreateFlag> flag, Progressable progress,
    3.     DataChecksum checksum, String[]favoredNodes, boolean createStreamer) {
    4.   this(dfsClient, src, flag, progress, stat,checksum);
    5.   this.shouldSyncBlock =flag.contains(CreateFlag.SYNC_BLOCK);
    6.  
    7.   // Directory => File => Block(128M)=> packet(64K) => chunk（chunk 512byte +chunksum 4byte）
    8.   computePacketChunkSize(dfsClient.getConf().getWritePacketSize(),
    9.       bytesPerChecksum);
    10.  
    11.   if (createStreamer) {
    12.     streamer = new DataStreamer(stat, null,dfsClient, src, progress,
    13.         checksum, cachingStrategy, byteArrayManager,favoredNodes,
    14.         addBlockFlags);
    15.   }
    16. }

    点击newStreamForCreate方法中的out.start()，进入DFSOutputStream.java

    1. protected synchronized void start() {
    2.        getStreamer().start();
    3. }
    4.  
    5. protected DataStreamer getStreamer() {
    6.        return streamer;
    7. }

    点击DataStreamer，进入DataStreamer.java

    1. class DataStreamer extends Daemon {
    2.     。。。。。。
    3. }

    点击Daemon，进入Daemon.java

    1. public class Daemon extends Thread {
    2.     。。。。。。
    3. }

    说明：out.start();实际是开启线程，点击DataStreamer，搜索run方法

    DataStreamer.java

    1. @Override
    2. public void run(){
    3.  
    4.        long lastPacket = Time.monotonicNow();
    5.        TraceScope scope = null;
    6.        while (!streamerClosed &&dfsClient.clientRunning) {
    7.         // if the Responder encountered an error, shutdown Responder
    8.         if (errorState.hasError()) {
    9.               closeResponder();
    10.          }
    11.  
    12.          DFSPacketone;
    13.         try {
    14.               // process datanode IO errors ifany
    15.               boolean doSleep =processDatanodeOrExternalError();
    16.  
    17.               final int halfSocketTimeout =dfsClient.getConf().getSocketTimeout()/2;
    18.               synchronized (dataQueue) {
    19.                // wait for a packet to be sent.
    20.                … …
    21.                      try {
    22.                        // 如果dataQueue里面没有数据，代码会阻塞在这儿
    23.                        dataQueue.wait(timeout);
    24.                      } catch(InterruptedException  e) {
    25.                        LOG.warn("Caught exception", e);
    26.                      }
    27.                      doSleep = false;
    28.                      now = Time.monotonicNow();
    29.                }
    30.                … …
    31.                      //  队列不为空，从队列中取出packet
    32.                      one = dataQueue.getFirst();// regular data packet
    33.                      SpanId[] parents =one.getTraceParents();
    34.                      if (parents.length > 0){
    35.                        scope = dfsClient.getTracer().
    36.                               newScope("dataStreamer",parents[0]);
    37.                        scope.getSpan().setParents(parents);
    38.                      }
    39.                }
    40.               }
    41.               … …
    42. }
  • 相关阅读:
    【FPGA零基础学习之旅#16】嵌入式块RAM-双口ram的使用
    【Javascript】创建对象的几种方式
    C++ 函数返回多个值的方法总结
    Axure药企内部管理平台+企业内部管理系统平台
    mysql创建数据库sql语句
    读取excel数据的方式整理
    MYSQL--事务
    windows11安装微软商店的ubuntu报错，已解决
    向openssl中添加一个最简单的算法
    Java使用Scanner类实现用户输入与交互
  • 原文地址：https://blog.csdn.net/zjjcchina/article/details/125991073