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 < nodes.length) {
 LOG.info("Excluding datanode " + nodes[errorIndex]);
 excludedNodes.add(nodes[errorIndex]);
 }
 // Connection failed. Let's wait a little bit and retry
 retry = true;
 }

 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 < 0) {
 throw new IOException("Mislabelled incoming datastream.");
 }
 DatanodeInfo targets[] = new DatanodeInfo[numTargets];
 for (int i = 0; i < targets.length; i++) {
 DatanodeInfo tmp = new DatanodeInfo();
 tmp.readFields(in);
 targets[i] = tmp;
 }
 Token accessToken = new Token();
 accessToken.readFields(in);