apache#6. datanode 写数据

Author: Yifan122

hadoop-common-project/hadoop-common/src/main/conf/core-site.xml

@@ -19,7 +19,7 @@
 <configuration>
     <property>
         <name>fs.defaultFS</name>
-        <value>hdfs://localhost:9000</value>
+        <value>hdfs://localhost/</value>
     </property>
 </configuration>
 

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/FSOutputSummer.java

@@ -160,6 +160,8 @@ protected synchronized int flushBuffer(boolean keep,
     int partialLen = bufLen % sum.getBytesPerChecksum();
     int lenToFlush = flushPartial ? bufLen : bufLen - partialLen;
     if (lenToFlush != 0) {
+      // TODO 核心的代码
+      // TODO HDFS文件 -》 block文件块 （128M） -> packet(64K) = 127chunk -> chunk (512  bytes) + chunk sum (4 bytes)
       writeChecksumChunks(buf, 0, lenToFlush);
       if (!flushPartial || keep) {
         count = partialLen;

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/DFSClient.java

@@ -1701,6 +1701,8 @@ public DFSOutputStream create(String src,
         src, masked, flag, createParent, replication, blockSize, progress,
         buffersize, dfsClientConf.createChecksum(checksumOpt),
         getFavoredNodesStr(favoredNodes));
+
+    // 开启契约管理
     beginFileLease(result.getFileId(), result);
     return result;
   }

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/DFSOutputStream.java

@@ -224,6 +224,14 @@ private DFSPacket createHeartbeatPacket() throws InterruptedIOException {
   // it. When all the packets for a block are sent out and acks for each
   // if them are received, the DataStreamer closes the current block.
   //
+
+  /**
+   * TODO
+   *  DataStreamer 的服务主要是用来处理接收数据管道里面的数据
+   *  它会向NameNode申请新的block，namenode返回来的时候，会返回关于block的blockid和block对应的DataNode的信息
+   *  启动了这个服务以后，它会接收packet流
+   *  每个packet流都有序列号，当这个服务收到有i个完整的block的时候，就会返回响应并关闭当前的DataStreamer
+   */
   class DataStreamer extends Daemon {
     private volatile boolean streamerClosed = false;
     private ExtendedBlock block; // its length is number of bytes acked
@@ -446,6 +454,12 @@ public void run() {
             if(DFSClient.LOG.isDebugEnabled()) {
               DFSClient.LOG.debug("Allocating new block");
             }
+            // TODO 步骤一：建立数据管道
+            /**
+             * nextBlockOutputStream 完成了两件事：
+             *  1）向NameNode申请block
+             *  2）建立数据管道
+             */
             setPipeline(nextBlockOutputStream());
             initDataStreaming();
           } else if (stage == BlockConstructionStage.PIPELINE_SETUP_APPEND) {
@@ -1223,6 +1237,7 @@ private LocatedBlock nextBlockOutputStream() throws IOException {
             .keySet()
             .toArray(new DatanodeInfo[0]);
         block = oldBlock;
+        // TODO 获得block
         lb = locateFollowingBlock(excluded.length > 0 ? excluded : null);
         block = lb.getBlock();
         block.setNumBytes(0);
@@ -1233,6 +1248,7 @@ private LocatedBlock nextBlockOutputStream() throws IOException {
 
         //
         // Connect to first DataNode in the list.
+        // TODO 链接到第一个datanode
         //
         success = createBlockOutputStream(nodes, storageTypes, 0L, false);
 
@@ -1281,10 +1297,13 @@ private boolean createBlockOutputStream(DatanodeInfo[] nodes,
         try {
           assert null == s : "Previous socket unclosed";
           assert null == blockReplyStream : "Previous blockReplyStream unclosed";
+          // 创建socket
           s = createSocketForPipeline(nodes[0], nodes.length, dfsClient);
           long writeTimeout = dfsClient.getDatanodeWriteTimeout(nodes.length);
-          
+
+          // 输出流：把数据写到datanode上
           OutputStream unbufOut = NetUtils.getOutputStream(s, writeTimeout);
+          // 输入流：读取响应的结果
           InputStream unbufIn = NetUtils.getInputStream(s);
           IOStreamPair saslStreams = dfsClient.saslClient.socketSend(s,
             unbufOut, unbufIn, dfsClient, accessToken, nodes[0]);
@@ -1307,13 +1326,18 @@ private boolean createBlockOutputStream(DatanodeInfo[] nodes,
 
           boolean[] targetPinnings = getPinnings(nodes, true);
           // send the request
+          // TODO 发送写数据的请求
+          // socket请求
+          // datanode会启动一个DataXceiverServer服务接受socket请求
+
           new Sender(out).writeBlock(blockCopy, nodeStorageTypes[0], accessToken,
               dfsClient.clientName, nodes, nodeStorageTypes, null, bcs, 
               nodes.length, block.getNumBytes(), bytesSent, newGS,
               checksum4WriteBlock, cachingStrategy.get(), isLazyPersistFile,
             (targetPinnings == null ? false : targetPinnings[0]), targetPinnings);
 
           // receive ack for connect
+          // TODO 等待datanode返回
           BlockOpResponseProto resp = BlockOpResponseProto.parseFrom(
               PBHelper.vintPrefixed(blockReplyStream));
           pipelineStatus = resp.getStatus();
@@ -1427,6 +1451,7 @@ private LocatedBlock locateFollowingBlock(DatanodeInfo[] excludedNodes)  throws
         long localstart = Time.monotonicNow();
         while (true) {
           try {
+            // TODO 获得写入的block
             return dfsClient.namenode.addBlock(src, dfsClient.clientName,
                 block, excludedNodes, fileId, favoredNodes);
           } catch (RemoteException e) {
@@ -1598,6 +1623,7 @@ private DFSOutputStream(DFSClient dfsClient, String src, HdfsFileStatus stat,
 
     computePacketChunkSize(dfsClient.getConf().writePacketSize, bytesPerChecksum);
 
+    // TODO 创建DataStreamer
     streamer = new DataStreamer(stat, null);
     if (favoredNodes != null && favoredNodes.length != 0) {
       streamer.setFavoredNodes(favoredNodes);
@@ -1620,6 +1646,11 @@ static DFSOutputStream newStreamForCreate(DFSClient dfsClient, String src,
       while (shouldRetry) {
         shouldRetry = false;
         try {
+          /**
+           * 上传文件：
+           *  1） 创建文件
+           *  2） 添加契约
+           */
           stat = dfsClient.namenode.create(src, masked, dfsClient.clientName,
               new EnumSetWritable<CreateFlag>(flag), createParent, replication,
               blockSize, SUPPORTED_CRYPTO_VERSIONS);
@@ -1651,6 +1682,7 @@ static DFSOutputStream newStreamForCreate(DFSClient dfsClient, String src,
         }
       }
       Preconditions.checkNotNull(stat, "HdfsFileStatus should not be null!");
+      // TODO 创建文件流
       final DFSOutputStream out = new DFSOutputStream(dfsClient, src, stat,
           flag, progress, checksum, favoredNodes);
       out.start();
@@ -1752,6 +1784,7 @@ private void waitAndQueueCurrentPacket() throws IOException {
               firstWait = false;
             }
             try {
+              // 如果队列写满了，就等待
               dataQueue.wait();
             } catch (InterruptedException e) {
               // If we get interrupted while waiting to queue data, we still need to get rid
@@ -1772,6 +1805,7 @@ private void waitAndQueueCurrentPacket() throws IOException {
           }
         }
         checkClosed();
+        // TODO 往队列里面添加packet
         queueCurrentPacket();
       } catch (ClosedChannelException e) {
       }
@@ -1807,6 +1841,7 @@ private synchronized void writeChunkImpl(byte[] b, int offset, int len,
     }
 
     if (currentPacket == null) {
+      // TODO 创建packet
       currentPacket = createPacket(packetSize, chunksPerPacket, 
           bytesCurBlock, currentSeqno++, false);
       if (DFSClient.LOG.isDebugEnabled()) {
@@ -1818,14 +1853,20 @@ private synchronized void writeChunkImpl(byte[] b, int offset, int len,
             ", bytesCurBlock=" + bytesCurBlock);
       }
     }
-
+    // TODO 往packet 里面写chunk的校验合 4bytes
     currentPacket.writeChecksum(checksum, ckoff, cklen);
     currentPacket.writeData(b, offset, len);
     currentPacket.incNumChunks();
+    // 达到128M 就是一个block
     bytesCurBlock += len;
 
     // If packet is full, enqueue it for transmission
     //
+    /**
+     * TODO 两个条件
+     *  1。 如果写满一个packet （127 chunk） = 一个packet
+     *  2。 如果写满了一个Block （128M） 2048packet
+     */
     if (currentPacket.getNumChunks() == currentPacket.getMaxChunks() ||
         bytesCurBlock == blockSize) {
       if (DFSClient.LOG.isDebugEnabled()) {

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/DistributedFileSystem.java

@@ -445,6 +445,11 @@ public FSDataOutputStream create(final Path f, final FsPermission permission,
       @Override
       public FSDataOutputStream doCall(final Path p)
           throws IOException, UnresolvedLinkException {
+        /**
+         * 1）往文件目录树里面添加了INodeFile
+         * 2）添加契约管理
+         * 3）启动了DataStreamer （写数据流程的关键服务）
+         */
         final DFSOutputStream dfsos = dfs.create(getPathName(p), permission,
                 cflags, replication, blockSize, progress, bufferSize,
                 checksumOpt);

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/DataXceiverServer.java

@@ -132,6 +132,7 @@ public void run() {
     Peer peer = null;
     while (datanode.shouldRun && !datanode.shutdownForUpgrade) {
       try {
+        // TODO 接受socket 请求
         peer = peerServer.accept();
 
         // Make sure the xceiver count is not exceeded

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSNamesystem.java

@@ -2436,6 +2436,7 @@ private HdfsFileStatus startFileInt(final String srcArg,
       try {
         src = dir.resolvePath(pc, src, pathComponents);
         final INodesInPath iip = dir.getINodesInPath4Write(src);
+        // TODO 重要代码
         toRemoveBlocks = startFileInternal(
             pc, iip, permissions, holder,
             clientMachine, create, overwrite,
@@ -2561,6 +2562,8 @@ private BlocksMapUpdateInfo startFileInternal(FSPermissionChecker pc,
       Map.Entry<INodesInPath, String> parent = FSDirMkdirOp
           .createAncestorDirectories(dir, iip, permissions);
       if (parent != null) {
+        // TODO 往文件目录树里面添加INodeFile节点
+        // dir就是FSDirectory
         iip = dir.addFile(parent.getKey(), parent.getValue(), permissions,
             replication, blockSize, holder, clientMachine);
         newNode = iip != null ? iip.getLastINode().asFile() : null;
@@ -2569,6 +2572,9 @@ private BlocksMapUpdateInfo startFileInternal(FSPermissionChecker pc,
       if (newNode == null) {
         throw new IOException("Unable to add " + src +  " to namespace");
       }
+      // TODO 添加契约
+      // 只有拥有契约的客户端才可以上传数据
+      // 契约会每隔一段时间更新
       leaseManager.addLease(newNode.getFileUnderConstructionFeature()
           .getClientName(), src);
 
@@ -3074,6 +3080,8 @@ LocatedBlock getAdditionalBlock(String src, long fileId, String clientName,
     }
 
     // choose targets for the new block to be allocated.
+    // TODO 选择存放block的datanode主机 （负载均衡）
+    // HDFS 机架感知
     final DatanodeStorageInfo targets[] = getBlockManager().chooseTarget4NewBlock( 
         src, replication, clientNode, excludedNodes, blockSize, favoredNodes,
         storagePolicyID);

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/NameNodeRpcServer.java

@@ -619,6 +619,7 @@ public HdfsFileStatus create(String src, FsPermission masked,
     try {
       PermissionStatus perm = new PermissionStatus(getRemoteUser()
           .getShortUserName(), null, masked);
+      // TODO 核心步骤
       status = namesystem.startFile(src, perm, clientName, clientMachine,
           flag.get(), createParent, replication, blockSize, supportedVersions,
           cacheEntry != null);
@@ -719,6 +720,12 @@ public LocatedBlock addBlock(String src, String clientName,
     }
     List<String> favoredNodesList = (favoredNodes == null) ? null
         : Arrays.asList(favoredNodes);
+
+      /**
+       * 1） 选择三台dataNode副本机器
+       * 2） 修改了目录树（内存）
+       * 3）存储元数据信息（磁盘）
+       */
     LocatedBlock locatedBlock = namesystem.getAdditionalBlock(src, fileId,
         clientName, previous, excludedNodesSet, favoredNodesList);
     if (locatedBlock != null)

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/TransferFsImage.java

@@ -240,6 +240,9 @@ public static void uploadImageFromStorage(URL fsName, Configuration conf,
 
   /*
    * Uploads the imagefile using HTTP PUT method
+   *
+   * 10PB的数据
+   * fsimage大概的大小： 1025 * 10 * 1024 = 4G
    */
   private static void uploadImage(URL url, Configuration conf,
       NNStorage storage, NameNodeFile nnf, long txId, Canceler canceler)

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/ha/StandbyCheckpointer.java

@@ -57,6 +57,12 @@
  * periodically waking up to take a checkpoint of the namespace.
  * When it takes a checkpoint, it saves it to its local
  * storage and then uploads it to the remote NameNode.
+ *
+ * 命名空间 = 元数据信息 = 目录树 =fsimage
+ *
+ * StandbyCheckpointer 是运行在standBynamenode上的一个线程
+ * 会周期性的对命名空间做checkpoint的操作
+ * 并把这份数据上传到active namenode上
  */
 @InterfaceAudience.Private
 public class StandbyCheckpointer {
@@ -182,6 +188,7 @@ private void doCheckpoint() throws InterruptedException, IOException {
       } else {
         imageType = NameNodeFile.IMAGE;
       }
+      // TODO 保存fsimage
       img.saveNamespace(namesystem, imageType, canceler);
       txid = img.getStorage().getMostRecentCheckpointTxId();
       assert txid == thisCheckpointTxId : "expected to save checkpoint at txid=" +
@@ -199,6 +206,7 @@ private void doCheckpoint() throws InterruptedException, IOException {
     // Upload the saved checkpoint back to the active
     // Do this in a separate thread to avoid blocking transition to active
     // See HDFS-4816
+    // TODO 上传fsimage到active namenode
     ExecutorService executor =
         Executors.newSingleThreadExecutor(uploadThreadFactory);
     Future<Void> upload = executor.submit(new Callable<Void>() {
@@ -320,7 +328,9 @@ private void doWork() {
           }
 
           final long now = monotonicNow();
+          // checkpoint条件一：是否超过1000000 条日志没有被checkpoint
           final long uncheckpointed = countUncheckpointedTxns();
+          // checkpoint条件二：是否超过3600s没有进行checkpoint了
           final long secsSinceLast = (now - lastCheckpointTime) / 1000;
 
           boolean needCheckpoint = needRollbackCheckpoint;