当前位置: 首页>后端>正文

HDFSDATANODE数据传输详解

本文主要阐述datanode中一个socket连接接收字节流的构成,帮助datanode的接收与处理数据。注意hadoop版本为3.1.1。

写在前面

Datanode本质上也是TCPServer,一般的TCPServer接到客户端请求以后会分配一个线程处理,对于Datanode而言,这个线程可以叫做Op处理连接。每个OP连接会多次和客户端交互,中间涉及多种packet。

关于proto writeDelimitedTo方法

在整个处理流程中,会非常频繁的使用的到proto.writeDelimitedTo来传递相关proto,简单理解就是要写入proto时,写入总长度,再写入proto。读取proto时,先读取长度,在解析proto。

DataNode连接数据流说明

OpPacket的接收

Op连接第一个包总是来定义Op连接处理那种Op,例如读块op,写块op。这种包简单命名为OpPacket。Packet的结构如下,可以根据下图读取OpPacket。

[图片上传失败...(image-766254-1710637927264)]

DataTransferProtocol.DATA_TRANSFER_VERSION:short,3.1.1版本默认为28。

OpCode:

[图片上传失败...(image-1c4ae9-1710637927264)]

OpProto:定义Op是哪种Op。Op在datatransfer.proto定义,包含OpReadBlockProto,OpWriteBlockProto,OpTransferBlockProto,OpReplaceBlockProto,OpCopyBlockProto,OpBlockChecksumProto,OpRequestShortCircuitAccessProto,ReleaseShortCircuitAccessRequestProto。

接下来的是否回应或者直接发数据,都要根据不同的op来处理,后续介绍了write和read

WriteBlock

当接收完OpPacket以后,需要写入一个BlockOpResponseProto应答。

[图片上传失败...(image-caf053-1710637927264)]

当客户端接受BlockOpResponseProto应答后,就会发送数据包,数据包的格式如下

[图片上传失败...(image-10ec21-1710637927264)]

PktLen:数据包长度,不同于字面意思,这个数值并不是包的总长度,而是4(pktLen所占字节数)+chunkchecksums字节数+chunkdatas字节数。
HeadLen:short,PacketHeaderProto的长度。不同于writeDelimitedTo,这边使用的proto.getSerializedSize。
PktHeadProto:PacketHeaderProto.writeTo。
Chunkchecksums:chunk校验数据。
ChunkData:实际数据。
DataNode接受到数据以后,完成checksum后就把Status.success放入Responder的处理队列。Responder最终会返回PipelineAckProto(PipelineAckProto.writeDelimitedTo)给客户端。

ReadBlock

当接收完OpPacket以后,需要写入一个BlockOpResponseProto应答。

[图片上传失败...(image-676e81-1710637927264)]

写完应答以后,立马会发送Block的数据包,数据包的结果如下:

[图片上传失败...(image-42c8ee-1710637927264)]

PktLen:数据包长度,不同于字面意思,这个数值并不是包的总长度,而是4(pktLen所占字节数)+chunkchecksums字节数+chunkdatas字节数。
HeadLen:short,PacketHeaderProto的长度。不同于writeDelimitedTo,这边使用的proto.getSerializedSize。
PktHeadProto:PacketHeaderProto.writeTo。
Chunkchecksums:chunk校验数据。
ChunkData:实际数据。
数据会被分成多个数据包发送,发送完最后一个数据包以后,会发送一个空包(没有数据只有header),空包的PacketHeaderProto会有LastPackctInBlock的标记。空包发送完成后,会接受一个ClientReadStatusProto的包(客户端使用ClientReadStatusProto.writeDelimitedTo写入)。

TransferBlock、ReplaceBlock未分析。

数据流中的sasl

Hadoop使用dfs.data.transfer.protection参数来保证数据流的安全。dfs.data.transfer.protection有三种模式authentication,integrity,privacy,分别对于sasl qop中的auth,auth-int,auth-conf。Auth:流建立需要握手,握手成功以后,后续流就是正常使用。
Auth-int:流握手+后续流都需要通过sasl的wrap unwarp加密解密。
Auth-conf:流握手+后续流都需要通过协商的算法来数据加密解密。

Sasl握手:
Sasl的mech为DIGEST-MD5,serviceName为0,protocol(c中的username)为hdfs。通过此信息可以创建saslclient,saslserver。DIGEST-MD5的callback的本质上就是验证用户名密码。数据流的用户密码来源与blocktoken。
关于sasl中协商的包结构为DataTransferEncryptorMessageProto,写入使用writeDelimitedTo。

message DataTransferEncryptorMessageProto {
  enum DataTransferEncryptorStatus {
    SUCCESS = 0;
    ERROR_UNKNOWN_KEY = 1;
    ERROR = 2;
  }
  required DataTransferEncryptorStatus status = 1;
  optional bytes payload = 2;
  optional string message = 3;
  repeated CipherOptionProto cipherOption = 4;
}

Payload就是token,message只有status为error才使用,为errmsg。Server发生异常都会发送错误,并关闭这个流。

流程图:

[图片上传失败...(image-a992ff-1710637927264)]

  1. client发送sasl_Version,为4byte

SASL_TRANSFER_MAGIC_NUMBER = 0xDEADBEEF;server接收并验证。

  1. client发送第一个saslMessage,Status为success,payload为byte[0]。

  2. Server就收到包以后使用saslserver.evaluateResponse(c中为gsasl_setup)来处理payload。

Server发送应答saslMessage,Status为success,payload为算出来的token。

  1. client接收到包以后,使用saslclient.evaluateChallenge(c中为gsasl_setup)来出来payload。

client发送第二个saslMessage,Status为success,payload为算出来的token。

  1. Server就收到包以后使用saslserver.evaluateResponse(c中为gsasl_setup)来处理payload。

这时候如果payload没问题,saslserver会complete。Server发送应答saslMessage,Status为success,payload为算出来的token。

  1. client接收到包以后,使用saslclient.evaluateChallenge(c中为gsasl_setup)来出来payload。

这时候如果payload没问题,saslclient会complete。


https://www.xamrdz.com/backend/3m71933871.html

相关文章: