HDFS Append时packet的格式以及DataNode对block/checksum文件的处理

451次阅读

共计 3271 个字符，预计需要花费 9 分钟才能阅读完成。

HDFS Append 时 packet 的格式以及 DataNode 对 block/checksum 文件的处理

HDFS 的 Block 一般比较大，默认 64MB/128MB，客户端给 DataNode 发数据实际上是以 Packet 的形式发送的，Packet 一般只有 64KB 左右。Packet 内部由分为一个个 chunk，每个 chunk 一般情况 (后面会说非一般情况) 下 512Bytes，并且 Packet 内部对于每个 chunk 会带上这个 chunk 的 checksum，对于 CRC 来说，checksum 本身占用 4Bytes。一个 Packet 的结构如下:
HDFS Append 时 packet 的格式以及 DataNode 对 block/checksum 文件的处理

其中 header 中包含了一些元信息，比如这个 packet 是不是所属 block 的最后一个 packet，数据长度多少，编码多少，packet 的数据部分的第一个字节在 block 中的 offset，DataNode 接到这个 Packet 是否必须 sync 磁盘。

下面主要关注当客户端使用如下模式写数据时，数据是怎样被封装成一个个 packet 的:

FSDataOutputStream out = fs.append(new Path(file));
out.write(b, off, len);
out.hflush()

首先，假设需要进行 append 的 file 一开始不存在，没有任何数据。
实际上，out 是 HdfsDataOutputStream 类型，out.write(b,off,len)最终调用了 HdfsDataOutputStream 包含的 DFSOutputStream 类型对象，转而调用的是 DFSOutputStream 的父类 FSOutputSummer 的 write(b,off,len)，FSOutputSummer 从名字可以看出，对数据进行了 checksum.

FSOutputSummer 的 write(b,off,len)实质上就一行:

for (int n=0;n<len;n+=write1(b, off+n, len-n)) {}

实际上就是循环的调用 write1 方法将数据写入，write1 方法会去检查是否写入的数据满了一个 chunk(正常情况下 512Bytes)，如果满了，则为这个 chunk 计算一个 checksum，4 个字节，然后将这个 chunk 和对应的 checksum 写入当前 Packet 中(DFSOutputStream 的 writeChunk 方法)，格式就是上面那个图中的格式。当 Packet 满了，也就是说塞入的 chunk 的个数到达了预先计算的值，就将这个 packet 放入 dataQueue，后台会有一个 DataStreamer 线程专门从这个 dataQueue 中取一个个的 packet 发送出去。

到这里，都比较清晰，需要注意的是，如果 append 的 file 中本身已经存在了一些数据，比如 512+512+100Bytes，那么在调用 out.write(b,off,len)向 file 中追加新的数据时，构造的第一个 packet 中只有一个 chunk，并且这个 chunk 的大小是 512-100=412Bytes，这个 packet 之后的 packet 的 chunk 是标准 512Bytes。

if (appendChunk && bytesCurBlock%bytesPerChecksum == 0) {appendChunk = false;
        // 下次计算 checksum 时的 chunk 大小是512Bytes，不再是412Bytes
        resetChecksumChunk(bytesPerChecksum);
}

以上代码来自 DFSOutputStream 的 writeChunk 方法，其中 appendChunk 在 fs.append 时会被置为 true，并且 bytesCurBlock 会初始化为 append 之前文件的大小，当构造完特殊包后，bytesCurBlock 增加了 412Bytes，将上次的 paritial chunk 补齐了。

DataStreamer 从 dataQueue 中取 packet 发送出去的过程不关注，下面看 DataNode 针对 append 如何处理 block 文件和 block 的 checksum 文件。

DataNode 上接受 Block 的逻辑封装在 BlockReceiver 中，其中,receiveBlock 方法中有一段代码

while (receivePacket() >= 0) {/* Receive until the last packet */ }

每调一次 receivePacket，DataNode 就接收一个 packet，将 packet 的 data 和 checksum 都拿出来，进行一次校验，看数据在发送过程中是否损坏，然后看 block 在当前 DataNode 上的副本在磁盘上的长度是否是 chunk 的整数倍，如果不是，则将 block checksum 文件 (blk_1100893627_27540491.meta) 的输出流 seek 到最后一个 checksum，并且将最后一个 checksum 读出来，如下代码：

 if (onDiskLen % bytesPerChecksum != 0) {// prepare to overwrite last checksum
            adjustCrcFilePosition();
 }
          
 // If this is a partial chunk, then read in pre-existing checksum
 if (firstByteInBlock % bytesPerChecksum != 0) {LOG.info("Packet starts at" + firstByteInBlock +
                     "for" + block +
                     "which is not a multiple of bytesPerChecksum" +
                     bytesPerChecksum);
   long offsetInChecksum = BlockMetadataHeader.getHeaderSize() +
                onDiskLen / bytesPerChecksum * checksumSize;
   computePartialChunkCrc(onDiskLen, offsetInChecksum, bytesPerChecksum);
 }