HDFS写文件过程分析

     static String[] contents = new String[] {
          "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
          "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb",
          "cccccccccccccccccccccccccccccccccccccccccccccccccccccccccc",
          "dddddddddddddddddddddddddddddddd",
          "eeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeee",
     };
     public static void main(String[] args) {
          String file = "hdfs://h1:8020/data/test/test.log";
        Path path = new Path(file);
        Configuration conf = new Configuration();
        FileSystem fs = null;
        FSDataOutputStream output = null;
        try {
               fs = path.getFileSystem(conf);
               output = fs.create(path); // 创建文件
               for(String line : contents) { // 写入数据
                    output.write(line.getBytes("UTF-8"));
                    output.flush();
               }
          } catch (IOException e) {
               e.printStackTrace();
          } finally {
               try {
                    output.close();
               } catch (IOException e) {
                    e.printStackTrace();
               }
          }
     }

  public FSDataOutputStream create(Path f, FsPermission permission,
    boolean overwrite,
    int bufferSize, short replication, long blockSize,
    Progressable progress) throws IOException {
    statistics.incrementWriteOps(1);
    return new FSDataOutputStream
       (dfs.create(getPathName(f), permission, overwrite, true, replication, blockSize, progress, bufferSize), statistics);
  }

  public OutputStream create(String src,
                             FsPermission permission,
                             boolean overwrite,
                             boolean createParent,
                             short replication,
                             long blockSize,
                             Progressable progress,
                             int buffersize
                             ) throws IOException {
   ... ...
}

    final DFSOutputStream result = new DFSOutputStream(src, masked,
        overwrite, createParent, replication, blockSize, progress, buffersize,
        conf.getInt("io.bytes.per.checksum", 512));

    DFSOutputStream(String src, FsPermission masked, boolean overwrite,
        boolean createParent, short replication, long blockSize, Progressable progress,
        int buffersize, int bytesPerChecksum) throws IOException {
      this(src, blockSize, progress, bytesPerChecksum, replication);
      computePacketChunkSize(writePacketSize, bytesPerChecksum); // 默认 writePacketSize=64*1024（即64K），bytesPerChecksum=512（没512个字节计算一个校验和）,
      try {
        if (createParent) { // createParent为true表示，如果待创建的文件的父级目录不存在，则自动创建
          namenode.create(src, masked, clientName, overwrite, replication, blockSize);
        } else {
          namenode.create(src, masked, clientName, overwrite, false, replication, blockSize);
        }
      } catch(RemoteException re) {
        throw re.unwrapRemoteException(AccessControlException.class,
                                       FileAlreadyExistsException.class,
                                       FileNotFoundException.class,
                                       NSQuotaExceededException.class,
                                       DSQuotaExceededException.class);
      }
      streamer.start(); // 启动一个DataStreamer线程，用来将写入的字节流打包成packet，然后发送到对应的Datanode节点上
    }
上面computePacketChunkSize方法计算了一个packet的相关参数，我们结合代码来查看，如下所示：
      int chunkSize = csize + checksum.getChecksumSize();
      int n = DataNode.PKT_HEADER_LEN + SIZE_OF_INTEGER;
      chunksPerPacket = Math.max((psize - n + chunkSize-1)/chunkSize, 1);
      packetSize = n + chunkSize*chunksPerPacket;

      int chunkSize = 512 + 4;
      int n = 21 + 4;
      chunksPerPacket = Math.max((64*1024 - 25 + 516-1)/516, 1);  // 127
      packetSize = 25 + 516*127;

        if (createParent) { // createParent为true表示，如果待创建的文件的父级目录不存在，则自动创建
          namenode.create(src, masked, clientName, overwrite, replication, blockSize);
        } else {
          namenode.create(src, masked, clientName, overwrite, false, replication, blockSize);
        }

      ByteBuffer buffer;           // only one of buf and buffer is non-null
      byte[]  buf;
      long    seqno;               // sequencenumber of buffer in block
      long    offsetInBlock;       // 该packet在block中的偏移量
      boolean lastPacketInBlock;   // is this the last packet in block?
      int     numChunks;           // number of chunks currently in packet
      int     maxChunks;           // 一个packet中包含的chunk的个数
      int     dataStart;
      int     dataPos;
      int     checksumStart;
      int     checksumPos; 

      Packet() {
        this.lastPacketInBlock = false;
        this.numChunks = 0;
        this.offsetInBlock = 0;
        this.seqno = HEART_BEAT_SEQNO; // 值为-1
        buffer = null;
        int packetSize = DataNode.PKT_HEADER_LEN + SIZE_OF_INTEGER; // 21+4=25
        buf = new byte[packetSize];
        checksumStart = dataStart = packetSize;
        checksumPos = checksumStart;
        dataPos = dataStart;
        maxChunks = 0;
      }

           Packet one = null;
          one = dataQueue.getFirst(); // regular data packet
          ByteBuffer buf = one.getBuffer();
          // write out data to remote datanode
          blockStream.write(buf.array(), buf.position(), buf.remaining());
          if (one.lastPacketInBlock) { // 如果是Block中的最后一个Packet，还要写入一个0标识该Block已经写入完成
              blockStream.writeInt(0); // indicate end-of-block
          }

        if (!success) {
          LOG.info("Abandoning " + block);
          namenode.abandonBlock(block, src, clientName);
          if (errorIndex 
<p><strong>DataNode端服务组件</strong></p>
<p>数据最终会发送到DataNode节点上，在一个DataNode上，数据在各个组件之间流动，流程如下图所示：<br>
<img alt="hdfs-write-pipeline-single" class="alignnone size-full wp-image-947"    style="max-width:90%" src="http://www.68idc.cn/help/uploads/allimg/150119/0RA42a0-4.png"  style="max-width:90%"><br>
DataNode服务中创建一个后台线程DataXceiverServer，它是一个SocketServer，用来接收来自Client（或者DataNode Pipeline中的非最后一个DataNode节点）的写数据请求，然后在DataXceiverServer中将连接过来的Socket直接派发给一个独立的后台线程DataXceiver进行处理。所以，Client写数据时连接一个DataNode Pipeline的结构，实际流程如图所示：<br>
<img alt="hdfs-write-pipeline-datanodes" class="alignnone size-full wp-image-948"    style="max-width:90%" src="http://www.68idc.cn/help/uploads/allimg/150119/0RA440a-5.png"  style="max-width:90%"><br>
每个DataNode服务中的DataXceiver后台线程接收到来自前一个节点（Client/DataNode）的Socket连接，首先读取Header数据：</p>
<pre class="brush:php;toolbar:false">
    Block block = new Block(in.readLong(), dataXceiverServer.estimateBlockSize, in.readLong());
    LOG.info("Receiving " + block + " src: " + remoteAddress + " dest: " + localAddress);
    int pipelineSize = in.readInt(); // num of datanodes in entire pipeline
    boolean isRecovery = in.readBoolean(); // is this part of recovery?
    String client = Text.readString(in); // working on behalf of this client
    boolean hasSrcDataNode = in.readBoolean(); // is src node info present
    if (hasSrcDataNode) {
      srcDataNode = new DatanodeInfo();
      srcDataNode.readFields(in);
    }
    int numTargets = in.readInt();
    if (numTargets  accessToken = new Token<blocktokenidentifier>();
    accessToken.readFields(in);
</blocktokenidentifier>