HDFS3源码分析之Client读流程(一)

云存储
, , ,

从前面背景介绍里, 可以知道HDFS核心由三大组件: Client, Server(Namenode), Worker(Datanode) 组成, 现在开始阅读它们的源码

因为Client部分最接近日常使用, 且相对清晰易懂, 所以总体按Client --> Namenode --> Datanode的流程来进行阅读, 核心围绕读写流程去把各个模块串联起来, 今天先来看HDFS读取文件的实现.

PS: 后面待重构, 因写入需要铺垫知识, 先放出来, 建议只看前面两小节.

0x00. 前置知识

A. 学习背景

HDFS整体的代码是非常多的, 加上文件系统本身有许多不得不啃的细节, 所以合理的源码阅读方式是必不可少的. 在读源码之前, 先说一下之后的基本参考资料和版本:

  • (源码)HDFS3.1.2 (主) 和 HDFS1.x/2.x (次): 读源码是必不可少的, 必须自己去读.

  • (博客)Cloudera-HDFS: 可以理解为HDFS官方的博客, 作者基本都是Commiter或PMC (网上大部分文章都是从这”摘抄”的)

  • (Slide)HDFS相关的SlideShare: 这也是一个很不错的途径来寻找相关高质量的HDFS分享, 推荐参考

  • (博客)深度剖析HDFS书作者博客: 少见的CSDN好博客, 作者很系统的写了HDFS不少知识点, 可供参考

  • (社区)HDFS-JIRA: 可以理解为HDFS社区, issue和patch(pr)的管理地 (建议装个patch阅读插件)

  • (书籍)HDFS2.X的源码剖析:

    书用做源码阅读辅助很好, 说下阅读顺序, 感觉读了第一章后, 差不多得逆序阅读, 然后穿插阅读, 最后读第二章RPC设计 (仅供参考)

之后Hadoop所有文章中提到的”官方”一般指Apache的hadoop官网, 和Cloudera官网, 也建议大家优先阅读官方文档和文章.

B. 基本概念 (重要)

在看代码读/写流程之前, 有一些基本的HDFS概念需要说明一下, 这样后面的细节才能更加清晰. (也可以之后再回头来看):

  1. block(文件块): 类似Linux中常见文件系统的最小操作单位, HDFS也是如此, block就是HDFS**操作文件的最小单元**, 默认**128MB/**个 (逻辑上设置的大小)
  2. chunk(校验块): 文件块(block)实际**存储/校验的最小单位, 严谨说chunk分为数据域和校验域两个部分组成, 但一般情况大家指的是数据域**
    • 数据域: 默认512字节, 比如你写513个字节数据, 它占据一个block (而此block内部是两个chunk组成) , 又称为chunk data
    • 校验域: 固定4字节, 存放校验码, 它与数据域一一对应, 又称chunk checksum
  3. packet(数据包): HDFS各组件间数据传输的基本单位, 默认64KB, 类似网络传输中数据包的概念, 主要是 header + body/data 两个部分组成
    • packet header : 存储传输数据过程中的一些基本信息(数据包长度/版本/标志位等), 变长
    • packet body: 存储实际传输的数据(chunk), 同样packet里实际分布的最小单位也是chunk

Packet理解最直观就是看看官方给的packet完整结构 (header+ data): (下图是我基于HDFS3修改过的版本, 与原图略有区别)

hdfsPacket00

然后再进一步, 每个HDFS上的文件(比如demo.txt)都有与之对应的block文件, 每个block都有与之唯一对应的meta文件存储校验信息 (包括校验版本/类型,本质是N个4字节的chunk校验域组成), 客户端用它来对比确定文件块是否完整. (如官方下图所示)

hdfsBlock00

所以这三个基本概念之间关系紧密可以理解为:

  • 1个block由多个chunk data组成, 每个chunk data里存储的都是二进制数据
  • 1个block对应有一个meta文件, 存放与之对应的chunk checksum
  • HDFS组件间传输数据最小以packet为单位, 1个packet由多个chunk构成, 传输后取出其中的chunk (data+checksum), 最后构成block和meta

总结: block是HDFS操作文件的最小单位, 它由chunk为单位组成, 但由packet为单位进行传输 , 三个概念都是读写过程中经常遇到的, 十分重要.

0x01. 读流程

这里说的读取流程就是指Client向Server发请求走网络读文件的过程, 网上流程图不少, 但是画的都略粗, 我稍微优化重画了一下(原图见此):

hdfs-read

先说明一下, 这个图相对书中的图做了一些调整, 补充说几点:

  • 我把Client和DFS(DistributedFileSystem)合为了一个, 因为DFS是中间状态, 觉得可以讲细节再说
  • RPC和SocketStream是HDFS中两种不同的通信方式, 并且发了几次RPC, 什么RPC是很重要的, 需要单独标识
  • 在第一步open()的时候, 这里普通读只发一次RPC给NN, 但严谨来说在读取的文件还没写完整的时候, 它还会给DN发RPC确认文件长度(图上没画, 但得说明)
  • 例子中文本有200MB, 所以它被分成了两个block, 它和数字组合可以理解成blockId (不连续)
  • 一个文件有多个block时, 会依次去读取, 并且如果读取某个block失败, 它会去尝试读取存储此block的其他DN (多个副本), 也需要体现出来

然后结合上图说一下整体流程:

  1. 客户端向namenode发RPC请求, 读取文件对应blocks的DN信息
  2. namenode检查文件是否存在，如果存在则获取文件的元信息（按顺序返回BlockId以及对应所在的datanode列表）
  3. 客户端收到block对应的DN信息后选取一个网络访问最近的DN, 依次读取每个数据块(客户端这里有校验逻辑, 如果发现异常会汇报坏块)
  4. 客户端与DN建立socket连接，通过流以packet为单位传输对应的数据块, 客户端收到数据写入本地磁盘
  5. 依次传输剩下的数据块，直到整个文件读取完成, 最后关闭输入流

对应实际的使用代码是:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
FileSystem fs = FileSystem.get(new URI("hdfs://ip:port"), new Configuration());
// 发RPC给NN,返回的stream对象包含文件对应block+dn列表 (假设文件已存在)
try (FSDataInputStream inStream = fs.open(new Path("/test/readFile"))) { // 自动close()
    int dataBytes = inStream.read();  // 读取核心在此
    
    // 然后若需要输出可以通过输出流显示 (细节暂略, 用工具类输出到控制台)
    IOUtils.copyBytes(inStream, System.out, 4096, false);
}  

/* 读核心 (调用DFSInputStream)
 * 依次读取第一,第二...第n个block网络最近的DN,读完校验.
 * 如果校验成功,则关闭与当前DN通信, 失败则标记(下次不会读它)并找下一个存有该块的DN(最多失败3次?)
 * 如果遍历block列表操作完成但文件还不是完整的,FS会再与NN通信获取剩余blockList
 */

下面就来分别看看打开文件对应的open() 和读取数据的read() ,按这两个函数分别展开, 形成一个完整的脉络

这里补充引用一张读代码调用的整体图, 觉得也不错, 大家可以参考一下:

hdfsReadRef00

0x03. 打开文件

先看一下打开文件 — open() 方法的整体调用时序:

sequenceDiagram 
FileSystem->> DistributedFileSystem: open(path)
DistributedFileSystem->> DFSClient: appendInternal()
DFSClient->>FSDataInputStream: appendFile()
FSDataInputStream-->>DistributedFileSystem: LocatedBlock
DistributedFileSystem -->> FileSystem: 1

Note right of lastLocatedBlock: 加了个Hdfs实现类

图待完善修正, 只是先放一下

0x03. 读取文件

  1. 读流程中选取最近的数据节点, 判断条件是(网络延迟? 机架距离? 随机?)
    • Namenode返回的数据块的存储位置是按照与客户端的距离远近排序返回的 (细节参考ClientContext.getNetworkDistance 方法)
  2. BlockReader接口(stream实现)
  3. PacketReceiver的doRead()

1.整体代码

读的方式有常见的几种, 对应入参:

  1. 顺序读 ()
  2. 随机读 (seek找给指定的位置)

核心在于DFSInputStream的 read()方法和client的校验 (鉴于几乎所有方法都抛IO异常, 大部分默认省去了)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
@Override
public synchronized int read() {
  if (oneByteBuf == null) oneByteBuf = new byte[1]; //init 1 byte
  
  int ret = read(oneByteBuf, 0, 1);
  return (ret <= 0) ? -1 : (oneByteBuf[0] & 0xff); // keep binary same
}

@Override
public synchronized int read(@Nonnull final byte buf[], int off, int len) {
  validatePositionedReadArgs(pos, buf, off, len);
  if (len == 0) return 0;
  
  // 将不同的BlockReader进行了封装, 真正读数据的是BlockReader
  ReaderStrategy byteArrayReader;
  byteArrayReader = new ByteArrayStrategy(buf, off, len, readStatistics, dfsClient);
  return readWithStrategy(byteArrayReader);
}

protected final Object infoLock = new Object(); // why use null Object as lockObj?
// 关键方法,省略部分检查
protected synchronized int readWithStrategy(ReaderStrategy strategy) {
  int len = strategy.getTargetLength();
  CorruptedBlocks corruptedBlocks = new CorruptedBlocks();
  failures = 0; // int type
  if (pos < getFileLength()) {
    int retries = 2; // 外层重试两次
    while (retries > 0) {
      try {
        // currentNode can be left as null if previous read had a checksum
        // error on the same block. See HDFS-3067
        if (pos > blockEnd || currentNode == null) {
          currentNode = blockSeekTo(pos);
        }
        int realLen = (int) Math.min(len, (blockEnd - pos + 1L));
        
        // 这锁目的主要是单纯建立一个同步代码块,而不限定当前对象持有(this)
        synchronized(infoLock) {
         if (locatedBlocks.isLastBlockComplete()) 
         realLen = (int) Math.min(realLen, locatedBlocks.getFileLength() - pos);
        }
        int result = readBuffer(strategy, realLen, corruptedBlocks);

        if (result >= 0) {
          pos += result;
        } else {
          // got a EOS from reader though we expect more data on it.
          throw new IOException("Unexpected EOS from the reader");
        }
        updateReadStatistics(readStatistics, result, blockReader);
        dfsClient.updateFileSystemReadStats(blockReader.getNetworkDistance(),
            result);
        return result;
      } catch (ChecksumException ce) { // 校验失败(文件不完整, 直接抛出异常, 不扣次数)
        throw ce;
      } catch (IOException e) {
        checkInterrupted(e);
        if (retries == 1) DFSClient.LOG.warn("DFS Read", e);
        blockEnd = -1;
        if (currentNode != null) addToDeadNodes(currentNode);
        if (--retries == 0) throw e;
      } finally {
       /* 判断是否需要通过ClientProtocol汇报块损坏 (不管最终读成功/校验失败)
        * 1. client读了多次,至少有1个成功,汇报已尝试的坏块
        * 2. 所有副本读失败,只汇报副本设置1的, 其他不会汇报
        *    因为官方考虑这种情况可能是client自己有问题(导致的读全失败)
        * Q: 全读失败的时候,不汇报有怎么进一步确定是谁的问题呢? 
        */
        reportCheckSumFailure(corruptedBlocks, 
                              getCurrentBlockLocationsLength(), false);
      }
    }
  }
  return -1;
}

2.0 读核心方法

  • blockSeekTo : 找到当前要读位置的块并为读做准备, 它包含三个功能:
    1. getBlockAt : 获取当前偏移对应的block的节点位置信息 (优先二分从缓存里查找)
    2. chooseDataNode: 从1中获取的列表里获取一个节点, 这里获取DN后, 最多试3次获取block, 所以最多应该可以尝试3次
    3. 实例化后面读数据的BlockReader对象 : 根据是否本地读创建对应的BlockReader实现类
  • readBuffer : 使用上面创建好的BlockReader对象进行读操作. 比如常用的BlockReaderRemote
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
private synchronized int readBuffer(ReaderStrategy reader, int len,
                                    CorruptedBlocks corruptedBlocks) {
    IOException ioe;

    /* 当前节点只尝试一次
     * Intention is to handle one common case of an error that is not a
     * failure on datanode or client : when DataNode closes the connection
     * since client is idle. If there are other cases of "non-errors" then
     * then a datanode might be retried by setting this to true again.
     */
    boolean retryCurrentNode = true;

    while (true) {
      // retry as many times as seekToNewSource allows.
      try {
        return reader.readFromBlock(blockReader, len);
      } catch (ChecksumException ce) {
        DFSClient.LOG.warn("Found Checksum error for "
            + getCurrentBlock() + " from " + currentNode
            + " at " + ce.getPos());
        ioe = ce;
        retryCurrentNode = false;
        // we want to remember which block replicas we have tried
        corruptedBlocks.addCorruptedBlock(getCurrentBlock(), currentNode);
      } catch (IOException e) {
        if (!retryCurrentNode) {
          DFSClient.LOG.warn("Exception while reading from "
              + getCurrentBlock() + " of " + src + " from "
              + currentNode, e);
        }
        ioe = e;
      }
      boolean sourceFound;
      if (retryCurrentNode) {
        /* possibly retry the same node so that transient errors don't
         * result in application level failures (e.g. Datanode could have
         * closed the connection because the client is idle for too long).
         */
        sourceFound = seekToBlockSource(pos);
      } else {
        addToDeadNodes(currentNode);
        sourceFound = seekToNewSource(pos);
      }
      if (!sourceFound) {
        throw ioe;
      }
      retryCurrentNode = false;
    }
  }

2.1 BlockReader实现

上面的DFS输入流read()方法核心是调用BlockReader接口的read()读取数据块返回字节数组, 这个接口根据读取类型不同有几个实现类, 今天主要关注的是走socket的网络读类BlockReaderRemote. (其他还有本地读, 外部块读)

优先级来看, 外部块读 > 本地读 > Unix Socket网络读 > TCP网络读 ,最后都返回一个BlockReader的实现.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
// 备注: BlockReaderRemote类是以前的RemoteBlockReader2改名而来   
  ByteBuffer curDataSlice = null;
  @Override
  public synchronized int read(byte[] buf, int off, int len){
      
    if (curDataSlice==null || curDataSlice.remaining()==0 && bytesNeededToFinish>0) {
      readNextPacket(); // 核心调用
    }

    // ByteBuffer读完了 (EOF)
    if (curDataSlice.remaining() == 0) return -1;

    int nRead = Math.min(curDataSlice.remaining(), len);
    curDataSlice.get(buf, off, nRead);

    return nRead;
  }

private void readNextPacket() throws IOException {
    //Read packet headers.
    packetReceiver.receiveNextPacket(in);

    PacketHeader curHeader = packetReceiver.getHeader();
    curDataSlice = packetReceiver.getDataSlice();
    
    assert curDataSlice.capacity() == curHeader.getDataLen();
    // Sanity check the lengths
    if (!curHeader.sanityCheck(lastSeqNo)) ("error in packet header "+curHeader);

    if (curHeader.getDataLen() > 0) {
      int chunks = 1 + (curHeader.getDataLen() - 1) / bytesPerChecksum;
      int checksumsLen = chunks * checksumSize;

      assert packetReceiver.getChecksumSlice().capacity() == checksumsLen;
      lastSeqNo = curHeader.getSeqno();
      if (verifyChecksum && curDataSlice.remaining() > 0) {
        // N.B.: the checksum error offset reported here is actually
        // relative to the start of the block, not the start of the file.
        // This is slightly misleading, but preserves the behavior from
        // the older BlockReader.
        // checkSum校验 (CRC32)
        checksum.verifyChunkedSums(curDataSlice,
                                   packetReceiver.getChecksumSlice(),
                                   filename, curHeader.getOffsetInBlock());
      }
      bytesNeededToFinish -= curHeader.getDataLen();
    }

    // First packet will include some data prior to the first byte
    // the user requested. Skip it.
    if (curHeader.getOffsetInBlock() < startOffset) {
      int newPos = (int) (startOffset - curHeader.getOffsetInBlock());
      curDataSlice.position(newPos);
    }

    // If we've now satisfied the whole client read, read one last packet header, which should be empty
    if (bytesNeededToFinish <= 0) {
      // 读尾部的空包?
      readTrailingEmptyPacket();
      // 读完成后, 给DN发响应码
      // 否则DN可能会主动关闭连接,尽管这不影响数据完整性
      if (verifyChecksum) {
        sendReadResult(Status.CHECKSUM_OK);
      } else {
        sendReadResult(Status.SUCCESS);
      }
    }
  }

3. PacketReceiver实现

这里是BlockReader的更底层实现, 核心方法doRead() .

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113

private ByteBuffer curPacketBuf = null;
private void doRead(ReadableByteChannel ch, InputStream in) {
    // Each packet looks like:
    //   PLEN    HLEN      HEADER     CHECKSUMS  DATA
    //   32-bit  16-bit   <protobuf>  <variable length>
    //
    // PLEN:      Payload长是算除头长外的总长度.
    // HLEN:      头长度
    //
    // HEADER:    真正的头数据信息(protobuf传输)
    // CHECKSUMS: crcs数据校验码4k一个,n个 (可选)
    // DATA       数据块
    
    curPacketBuf.clear();
    curPacketBuf.limit(PacketHeader.PKT_LENGTHS_LEN); // 6(2+4)
    doReadFully(ch, in, curPacketBuf);
    curPacketBuf.flip();
    int payloadLen = curPacketBuf.getInt();

    if (payloadLen < Ints.BYTES) {
      // The "payload length" includes its own length. Therefore it
      // should never be less than 4 bytes
      throw new IOException("Invalid payload length " + payloadLen);
    }
    int dataPlusChecksumLen = payloadLen - Ints.BYTES;
    int headerLen = curPacketBuf.getShort();
    if (headerLen < 0) throwExcept("Invalid header length " + headerLen);

    // Sanity check the buffer size so we don't allocate too much memory and OOME.
    int totalLen = payloadLen + headerLen;
    if (totalLen < 0 || totalLen > MAX_PACKET_SIZE) 
        throwExcept("Incorrect value for packet payload size: " +payloadLen);

    // Make sure we have space for the whole packet, and read it.
    reallocPacketBuf(PacketHeader.PKT_LENGTHS_LEN + 
                     dataPlusChecksumLen + headerLen);
    curPacketBuf.clear();
    curPacketBuf.position(PacketHeader.PKT_LENGTHS_LEN);
    curPacketBuf.limit(PacketHeader.PKT_LENGTHS_LEN + 
                       dataPlusChecksumLen + 
                       headerLen);
    doReadFully(ch, in, curPacketBuf);
    curPacketBuf.flip();
    curPacketBuf.position(PacketHeader.PKT_LENGTHS_LEN);

    // Extract the header from the front of the buffer (after the length prefixes)
    byte[] headerBuf = new byte[headerLen];
    curPacketBuf.get(headerBuf);
    
    // 初始化Packet头
    if (curHeader == null) curHeader = new PacketHeader();
    curHeader.setFieldsFromData(payloadLen, headerBuf);

    // 计算 sub-slices of the packet
    int checksumLen = dataPlusChecksumLen - curHeader.getDataLen();
    throwsExcepIfCheckSumLenLessThanZero(checksumLen);

    reslicePacket(headerLen, checksumLen, curHeader.getDataLen());
  }

// 全读
private static void doReadFully(ReadableByteChannel channel, InputStream in, 
                                ByteBuffer buf) {
    if (channel != null) {
      while (buf.remaining() > 0) {
      int n = channel.read(buf);
      if (n < 0) throwExcept("Premature EOF reading from " + channel);
    }

    } else {
      int toRead = buf.remaining();
      int off = buf.arrayOffset() + buf.position();
      while (toRead > 0) {
      int ret = in.read(buf.array(), off, toRead);
      if (ret < 0) throwExcept( "Premature EOF from inputStream");
      toRead = toRead - ret;
      off = off + ret;
    }
      buf.position(buf.position() + buf.remaining());
    }
  }

private void reslicePacket(
      int headerLen, int checksumsLen, int dataLen) {
    // Packet structure (refer to doRead() for details):
    //   PLEN    HLEN      HEADER     CHECKSUMS  DATA
    //   32-bit  16-bit   <protobuf>  <variable length>
    //   |--- lenThroughHeader ----|
    //   |----------- lenThroughChecksums   ----|
    //   |------------------- lenThroughData    ------|
    int lenThroughHeader = PacketHeader.PKT_LENGTHS_LEN + headerLen;
    int lenThroughChecksums = lenThroughHeader + checksumsLen;
    int lenThroughData = lenThroughChecksums + dataLen;

    assert curPacketBuf.position() == lenThroughHeader;
    assert curPacketBuf.limit() == lenThroughData;

    // Slice the checksums.
    curPacketBuf.position(lenThroughHeader);
    curPacketBuf.limit(lenThroughChecksums);
    curChecksumSlice = curPacketBuf.slice();

    // Slice the data.
    curPacketBuf.position(lenThroughChecksums);
    curPacketBuf.limit(lenThroughData);
    curDataSlice = curPacketBuf.slice();

    // Reset buffer to point to the entirety of the packet (including
    // length prefixes)
    curPacketBuf.position(0);
    curPacketBuf.limit(lenThroughData);
  }

0x04. QA

思考问题:

  1. 客户端保证传输过程数据完整性的校验? (blk*.meta文件里存了与chunk data相对应的校验码(checksum), 传输文件的时候从packet中可以得到每部分的校验码, 最后得到整个block的校验码, 也就可以在客户端取总和与meta做比较, 看这个block是否完整了)

  2. 这里默认的block大小是128M, 如果修改为256M. 之前的数据会有多少影响? (之前的没有影响, 新数据按新的blockSize, 旧的按之前的, 本质个人觉得是因为block只是一个抽象概念)

  3. 什么情况下, 某个block会丢失 ?

  4. 读某个节点的block报错, 几种可能?

    • 在获取block对应的节点前报错, 尝试3次后就会抛出异常

    • 在获取block对应的节点后, 读block时报错, 两种常见情况

      1. 校验码报错, 此时会换一个节点读

        • 如最终读成功,

        • 如果非单副本全失败, 则认为客户端异常, 抛出后结束

      2. IO异常, 比如直接断网读失败了, 会减1次尝试次数, 把失败节点放入deadNodes ,(读1个block维护一个deadNodes), 然后重试

备注 : 读流程在NameNode和DataNode的处理逻辑因为篇幅所限, 后续的笔记会单独再说. 也包括其他的几种特殊读(以及并行/优选读 )

参考资料:
  1. 官方文档及其衍生文档/图片.(英文教材/教程之类)
  2. Hadoop 2.X HDFS源码剖析-Chap.5.2.x
  3. Github-HDFS3.1.2-SourceCode
  4. (后续参考)hdfs读细节和断点分析