HDFS写文件过程分析

时间 2019-12-02 标签 hdfs 文件过程分析

  参考： 

  　　HDFS写文件过程分析http://shiyanjun.cn/archives/942.html 

  　　HDFS的工做流程分析https://blog.csdn.net/z66261123/article/details/51194204 

  　　简单搞定hdfs读写流程https://blog.csdn.net/github_36444580/article/details/77840481 

  　　Hadoop核心-HDFS读写流程https://yq.aliyun.com/articles/325428 

 
 HDFS的工做机制概述 

HDFS集群分为两大角色：NameNode、DataNode
NameNode负责管理整个文件系统的元数据
DataNode 负责管理用户的文件数据块
文件会按照固定的大小（blocksize）切成若干块后分布式存储在若干台datanode上
每个文件块能够有多个副本，并存放在不一样的datanode上
Datanode会按期向Namenode汇报自身所保存的文件block信息，而namenode则会负责保持文件的副本数量
HDFS的内部工做机制对客户端保持透明，客户端请求访问HDFS都是经过向namenode申请来进行

HDFS写文件过程分析html

　　HDFS是一个分布式文件系统，在HDFS上写文件的过程与咱们平时使用的单机文件系统很是不一样，从宏观上来看，在HDFS文件系统上建立并写一个文件，流程以下图（来自《Hadoop：The Definitive Guide》一书）所示：node

具体过程描述以下：git

Client调用DistributedFileSystem对象的create方法，建立一个文件输出流（FSDataOutputStream）对象
经过DistributedFileSystem对象与Hadoop集群的NameNode进行一次RPC远程调用，在HDFS的Namespace中建立一个文件条目（Entry），该条目没有任何的Block
经过FSDataOutputStream对象，向DataNode写入数据，数据首先被写入FSDataOutputStream对象内部的Buffer中，而后数据被分割成一个个Packet数据包
以Packet最小单位，基于Socket链接发送到按特定算法选择的HDFS集群中一组DataNode（正常是3个，可能大于等于1）中的一个节点上，在这组DataNode组成的Pipeline上依次传输Packet
这组DataNode组成的Pipeline反方向上，发送ack，最终由Pipeline中第一个DataNode节点将Pipeline ack发送给Client
完成向文件写入数据，Client在文件输出流（FSDataOutputStream）对象上调用close方法，关闭流
调用DistributedFileSystem对象的complete方法，通知NameNode文件写入成功

更详细的流程：github

client发起文件上传请求,经过RPC与NameNode创建链接,NameNode检查目标文件是否已经存在,父目录是否存在,并检查用户是否有相应的权限,若检查经过,会为该文件建立一个新的记录,不然的话文件建立失败,客户端获得异常信息
client经过请求NameNode,第一个block应该传输到哪些DataNode服务器上
NameNode根据配置文件中指定的备份(replica)数量及机架感知原理进行文件分配,返回可用的DataNode的地址以三台DataNode为例:A B C。注: Hadoop在设计时考虑到数据的安全与高效,数据文件默认在HDFS上存放三份,存储策略为:第一个备份放在客户端相同的datanode上(若客户端在集群外运行,就随机选取一个datanode来存放第一个replica),第二个replica放在与第一个replica不一样机架的一个随机datanode上,第三个replica放在与第二个replica相同机架的随机datanode上,若是replica数大于三,则随后的replica在集群中随机存放,Hadoop会尽可能避免过多的replica存放在同一个机架上.选取replica存放在同一个机架上.(Hadoop 1.x之后容许replica是可插拔的,意思是说能够定制本身须要的replica分配策略)
client请求3台的DataNode的一台A上传数据,(本质是一个RPC调用,创建pipeline),A收到请求会继续调用B,而后B调用C,将整个pipeline创建完成后,逐级返回client
client开始往A上传第一个block(先从磁盘读取数据放到一个本地内存缓存),以packet为单位(默认 64K),A收到一个packet就会传给B,B传递给C;A每传一个packet会放入一个应答队列等待应答。注: 若是某个datanode在写数据的时候宕掉了下面这些对用户透明的步骤会被执行:数据被分割成一个个packet数据包在pipeline上一次传输,在pipeline反方向上,逐个发送ack(命令正确应答),最终由pipeline中第一个DataNode节点A将pipeline ack发送给client
1. 管道线关闭,全部确认队列上的数据会被挪到数据队列的首部从新发送,这样也就确保管道线中宕掉的datanode下流的datanode不会由于宕掉的datanode而丢失数据包
2. 在还在正常运行datanode上的当前block上作一个标志,这样当宕掉的datanode从新启动之后namenode就会知道该datanode上哪一个block是刚才宕机残留下的局部损坏block,从而把他删除掉
3. 已经宕掉的datanode从管道线中被移除,未写完的block的其余数据继续呗写入到其余两个还在正常运行的datanode中,namenode知道这个block还处在under-replicated状态(即备份数不足的状态)下,而后它会安排一个新的replica从而达到要求的备份数,后续的block写入方法同前面正常时候同样
4. 有可能管道线中的多个datanode宕掉(通常这种状况不多),但只要dfs.relication.min(默认值为1)个replica被建立,我么就认为该建立成功了,剩余的relica会在之后异步建立以达到指定的replica数
当一个block传输完成后,client再次发送请求NameNode上传第二个block到服务器

机架感知（副本节点选择）：算法

第一个副本在client所处的节点上。若是客户端在集群外，随机选一个
第二个副本和第一个副本位于相同机架，随机节点。
第三个副本位于不一样机架，随机节点

下面代码使用Hadoop的API来实现向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();
          }
     }
}

结合上面的示例代码，咱们先从fs.create(path);开始，能够看到FileSystem的实现DistributedFileSystem中给出了最终返回FSDataOutputStream对象的抽象逻辑，代码以下所示：安全

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);
}

上面，DFSClient dfs的create方法中建立了一个OutputStream对象，在DFSClient的create方法：服务器

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

建立了一个DFSOutputStream对象，以下所示：

 
   final DFSOutputStream result = new DFSOutputStream(src, masked,
    overwrite, createParent, replication, blockSize, progress, buffersize,
    conf.getInt("io.bytes.per.checksum", 512)); 
  
下面，咱们从DFSOutputStream类开始，说明其内部实现原理。数据结构
 
   DFSOutputStream内部原理 
  
    　　打开一个DFSOutputStream流，Client会写数据到流内部的一个缓冲区中，而后数据被分解成多个Packet，每一个Packet大小为64k字节，每一个Packet又由一组chunk和这组chunk对应的checksum数据组成，默认chunk大小为512字节，每一个checksum是对512字节数据计算的校验和数据。 
  
    　　当Client写入的字节流数据达到一个Packet的长度，这个Packet会被构建出来，而后会被放到队列dataQueue中，接着DataStreamer线程会不断地从dataQueue队列中取出Packet，发送到复制Pipeline中的第一个DataNode上，并将该Packet从dataQueue队列中移到ackQueue队列中。 
  
    　　ResponseProcessor线程接收从Datanode发送过来的ack，若是是一个成功的ack，表示复制Pipeline中的全部Datanode都已经接收到这个Packet，ResponseProcessor线程将packet从队列ackQueue中删除。 
  
    　　在发送过程当中，若是发生错误，全部未完成的Packet都会从ackQueue队列中移除掉，而后从新建立一个新的Pipeline，排除掉出错的那些DataNode节点，接着DataStreamer线程继续从dataQueue队列中发送Packet。 
  
    　　下面是DFSOutputStream的结构及其原理，如图所示： 
  
    　　咱们从下面3个方面来描述内部流程： 
  
   建立Packet，Client写数据时，会将字节流数据缓存到内部的缓冲区中，当长度知足一个Chunk大小（512B）时，便会建立一个Packet对象，而后向该Packet对象中写Chunk Checksum校验和数据，以及实际数据块Chunk Data，校验和数据是基于实际数据块计算获得的。每次知足一个Chunk大小时，都会向Packet中写上述数据内容，直到达到一个Packet对象大小（64K），就会将该Packet对象放入到dataQueue队列中，等待DataStreamer线程取出并发送到DataNode节点。
发送Packet，DataStreamer线程从dataQueue队列中取出Packet对象，放到ackQueue队列中，而后向DataNode节点发送这个Packet对象所对应的数据
接收ack，发送一个Packet数据包之后，会有一个用来接收ack的ResponseProcessor线程，若是收到成功的ack，则表示一个Packet发送成功。若是成功，则ResponseProcessor线程会将ackQueue队列中对应的Packet删除
 
   DFSOutputStream初始化 
  
     首先看一下，DFSOutputStream的初始化过程，构造方法以下所示： 
   
         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; 
    
上面对应的参数，说明以下表所示：
 
       参数名称 
       参数值 
       参数含义 
      
       chunkSize 
       512+4=516 
       每一个chunk的字节数（数据+校验和） 
      
       csize 
       512 
       每一个chunk数据的字节数 
      
       psize 
       64*1024 
       每一个packet的最大字节数（不包含header） 
      
       DataNode.PKT_HEADER_LEN 
       21 
       每一个packet的header的字节数 
      
       chunksPerPacket 
       127 
       组成每一个packet的chunk的个数 
      
       packetSize 
       25+516*127=65557 
       每一个packet的字节数（一个header+一组chunk）

在计算好一个packet相关的参数之后，调用create方法与Namenode进行RPC请求，请求建立文件：

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

远程调用上面方法，会在FSNamesystem中建立对应的文件路径，并初始化与该建立的文件相关的一些信息，如租约（向Datanode节点写数据的凭据）。文件在FSNamesystem中建立成功，就要初始化并启动一个DataStreamer线程，用来向Datanode写数据，后面咱们详细说明具体处理逻辑。

Packet结构与定义

Client向HDFS写数据，数据会被组装成Packet，而后发送到Datanode节点。Packet分为两类，一类是实际数据包，另外一类是heatbeat包。一个Packet数据包的组成结构，如图所示：

上图中，一个Packet是由Header和Data两部分组成，其中Header部分包含了一个Packet的概要属性信息，以下表所示：

字段名称	字段类型	字段长度	字段含义
pktLen	int	4	4 + dataLen + checksumLen
offsetInBlock	long	8	Packet在Block中偏移量
seqNo	long	8	Packet序列号，在同一个Block惟一
lastPacketInBlock	boolean	1	是不是一个Block的最后一个Packet
dataLen	int	4	dataPos – dataStart，不包含Header和Checksum的长度

Data部分是一个Packet的实际数据部分，主要包括一个4字节校验和（Checksum）与一个Chunk部分，Chunk部分最大为512字节。

在构建一个Packet的过程当中，首先将字节流数据写入一个buffer缓冲区中，也就是从偏移量为25的位置（checksumStart）开始写Packet数据的Chunk Checksum部分，从偏移量为533的位置（dataStart）开始写Packet数据的Chunk Data部分，直到一个Packet建立完成为止。若是一个Packet的大小未能达到最大长度，也就是上图对应的缓冲区中，Chunk Checksum与Chunk Data之间还保留了一段未被写过的缓冲区位置，这种状况说明，已经在写一个文件的最后一个Block的最后一个Packet。在发送这个Packet以前，会检查Chunksum与Chunk Data之间的缓冲区是否为空白缓冲区（gap），若是有则将Chunk Data部分向前移动，使得Chunk Data 1与Chunk Checksum N相邻，而后才会被发送到DataNode节点。

咱们看一下Packet对应的Packet类定义，定义了以下一些字段：

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类有一个默认的没有参数的构造方法，它是用来作heatbeat的，以下所示：

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;
}

经过代码能够看到，一个heatbeat的内容，实际上只有一个长度为25字节的header数据。经过this.seqno = HEART_BEAT_SEQNO;的值能够判断一个packet是不是heatbeat包，若是seqno为-1表示这是一个heatbeat包。

Client发送Packet数据

能够DFSClient类中看到，发送一个Packet以前，首先须要向选定的DataNode发送一个Header数据包，代表要向DataNode写数据，该Header的数据结构，如图所示：

上图显示的是Client发送Packet到第一个DataNode节点的Header数据结构，主要包括待发送的Packet所在的Block（先向NameNode分配Block ID等信息）的相关信息、Pipeline中另外2个DataNode的信息、访问令牌（Access Token）和校验和信息，Header中各个字段及其类型，详见下表：

字段名称	字段类型	字段长度	字段含义
Transfer Version	short	2	Client与DataNode之间数据传输版本号，由常量DataTransferProtocol.DATA_TRANSFER_VERSION定义，值为17
OP	int	4	操做类型，由常量DataTransferProtocol.OP_WRITE_BLOCK定义，值为80
blkId	long	8	Block的ID值，由NameNode分配
GS	long	8	时间戳（Generation Stamp），NameNode分配blkId的时候生成的时间戳
DNCnt	int	4	DataNode复制Pipeline中DataNode节点的数量
Recovery Flag	boolean	1	Recover标志
Client	Text		Client主机的名称，在使用Text进行序列化的时候，实际包含长度len与主机名称字符串ClientHost
srcNode	boolean	1	是否发送src node的信息，默认值为false，不发送src node的信息
nonSrcDNCnt	int	4	由Client写的该Header数据，该数不包含Pipeline中第一个节点（即为DNCnt-1）
DN2	DatanodeInfo		DataNode信息，包括StorageID、InfoPort、IpcPort、capacity、DfsUsed、remaining、LastUpdate、XceiverCount、Location、HostName、AdminState
DN3	DatanodeInfo		DataNode信息，包括StorageID、InfoPort、IpcPort、capacity、DfsUsed、remaining、LastUpdate、XceiverCount、Location、HostName、AdminState
Access Token	Token		访问令牌信息，包括IdentifierLength、Identifier、PwdLength、Pwd、KindLength、Kind、ServiceLength、Service
CheckSum Header	DataChecksum	1+4	校验和Header信息，包括type、bytesPerChecksum

Header数据包发送成功，Client会收到一个成功响应码（DataTransferProtocol.OP_STATUS_SUCCESS = 0），接着将Packet数据发送到Pipeline中第一个DataNode上，以下所示：

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
}

不然，若是失败，则会与NameNode进行RPC调用，删除该Block，并把该Pipeline中第一个DataNode加入到excludedNodes列表中，代码以下所示：

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;
}

DataNode端服务组件

数据最终会发送到DataNode节点上，在一个DataNode上，数据在各个组件之间流动，流程以下图所示：

DataNode服务中建立一个后台线程DataXceiverServer，它是一个SocketServer，用来接收来自Client（或者DataNode Pipeline中的非最后一个DataNode节点）的写数据请求，而后在DataXceiverServer中将链接过来的Socket直接派发给一个独立的后台线程DataXceiver进行处理。因此，Client写数据时链接一个DataNode Pipeline的结构，实际流程如图所示：

每一个DataNode服务中的DataXceiver后台线程接收到来自前一个节点（Client/DataNode）的Socket链接，首先读取Header数据：

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<BlockTokenIdentifier> accessToken = new Token<BlockTokenIdentifier>();
accessToken.readFields(in);

上面代码中，读取Header的数据，与前一个Client/DataNode写入Header字段的顺序相对应，再也不累述。在完成读取Header数据后，当前DataNode会首先将Header数据再发送到Pipeline中下一个DataNode结点，固然该DataNode确定不是Pipeline中最后一个DataNode节点。接着，该DataNode会接收来自前一个Client/DataNode节点发送的Packet数据，接收Packet数据的逻辑实际上在BlockReceiver中完成，包括未来自前一个Client/DataNode节点发送的Packet数据写入本地磁盘。在BlockReceiver中，首先会将接收到的Packet数据发送写入到Pipeline中下一个DataNode节点，而后再将接收到的数据写入到本地磁盘的Block文件中。

DataNode持久化Packet数据

在DataNode节点的BlockReceiver中进行Packet数据的持久化，一个Packet是一个Block中一个数据分组，咱们首先看一下，一个Block在持久化到磁盘上的物理存储结构，以下图所示：

每一个Block文件（如上图中blk_1084013198文件）都对应一个meta文件（如上图中blk_1084013198_10273532.meta文件），Block文件是一个一个Chunk的二进制数据（每一个Chunk的大小是512字节），而meta文件是与每个Chunk对应的Checksum数据，是序列化形式存储。

写文件过程当中Client/DataNode与NameNode进行RPC调用

Client在HDFS文件系统中写文件过程当中，会发生屡次与NameNode节点进行RPC调用来完成写数据相关操做，主要是在以下时机进行RPC调用：

写文件开始时建立文件：Client调用create在NameNode节点的Namespace中建立一个标识该文件的条目
在Client链接Pipeline中第一个DataNode节点以前，Client调用addBlock分配一个Block（blkId+DataNode列表+租约）
若是与Pipeline中第一个DataNode节点链接失败，Client调用abandonBlock放弃一个已经分配的Block
一个Block已经写入到DataNode节点磁盘，Client调用fsync让NameNode持久化Block的位置信息数据
文件写完之后，Client调用complete方法通知NameNode写入文件成功
DataNode节点接收到并成功持久化一个Block的数据后，DataNode调用blockReceived方法通知NameNode已经接收到Block

参数名称	参数值	参数含义
chunkSize	512+4=516	每一个chunk的字节数（数据+校验和）
csize	512	每一个chunk数据的字节数
psize	64*1024	每一个packet的最大字节数（不包含header）
DataNode.PKT_HEADER_LEN	21	每一个packet的header的字节数
chunksPerPacket	127	组成每一个packet的chunk的个数
packetSize	25+516*127=65557	每一个packet的字节数（一个header+一组chunk）