Hadoop Client 與 DataNode 通信協議深度解析：Read/Write Block 流程與數據格式

架構概覽

• Client ↔ NameNode：RPC 協議（元數據操作）
• Client ↔ DataNode：直接 Socket 通信（數據傳輸），非 RPC
• 版本差異：0.20.2/0.19 協議一致；0.23+ 引入 Protobuf 序列化

Write Block 流程（Pipeline 寫入）

1. Client 調用 NameNode.create() 請求創建文件，獲取 DataNode 列表
2. 啟動 DataStreamer 線程，主線程讀取本地數據封裝為 Packet 放入 dataQueue
3. DataStreamer 從隊列取 Packet，建立 BlockOutputStream 連接首個 DataNode，發送 DATA_TRANSFER_HEADER
4. 啟動 ResponseProcessor 線程接收 ACK 確認
5. Pipeline 方式將數據依次推送到下游 DataNode，並反向傳回 ACK

Write 數據包格式

// Header 結構
DataTransferProtocol.DATA_TRANSFER_VERSION  // 版本號
DataTransferProtocol.OP_WRITE_BLOCK         // 操作類型
blockId, generationStamp, ...               // 元數據

// Packet 結構
packetLen, offsetInBlock, seqno, lastPacketInBlock, dataLen
dataBytes...
checksumBytes (CRC32, 每 512 字節一個 checksum)

Read Block 流程

核心類：BlockReader（Hadoop 2.x 為 RemoteBlockReader2）

初始化階段

// 1. 建立 Socket 連接
Socket sock = new Socket(dnAddr, dnPort);
DataOutputStream out = new DataOutputStream(sock.getOutputStream());

// 2. 發送 Read Request Header
out.writeShort(DataTransferProtocol.DATA_TRANSFER_VERSION);
out.write(DataTransferProtocol.OP_READ_BLOCK);
out.writeLong(blockId);
out.writeLong(genStamp);
out.writeLong(startOffset);
out.writeLong(len);
Text.writeString(out, clientName);
out.flush();

// 3. 讀取 Response Header
DataInputStream in = new DataInputStream(sock.getInputStream());
short status = in.readShort();  // OP_STATUS_SUCCESS = 0

// 4. 讀取 Checksum 信息
DataChecksum checksum = DataChecksum.newDataChecksum(in);
// 默認：512 字節/Chunk，4 字節 CRC32

// 5. 讀取首個 Chunk 偏移量
long firstChunkOffset = in.readLong();

數據讀取循環

while (bytesNeeded > 0) {
    int packetLen = in.readInt();
    long offsetInBlock = in.readLong();
    long seqno = in.readLong();
    boolean lastPacketInBlock = in.readBoolean();
    int dataLen = in.readInt();
    
    // 讀取 checksum
    IOUtils.readFully(in, checksumBytes.array(), 0, checksumBytes.limit());
    
    // 讀取實際數據
    IOUtils.readFully(in, buf, offset, chunkLen);
    
    // 校驗 checksum
    FSInputChecker.verifySum(chunkPos);
}

關鍵設計點

• Checksum 校驗：每 512 字節計算 4 字節 CRC32，網絡傳輸中檢測數據損壞
• 流式處理：不將整個 Block 載入內存，邊讀邊校驗邊交付上層
• 錯誤恢復：校驗失敗時向 NameNode 報告壞塊，嘗試從副本讀取
• 協議版本演進：0.23+ 使用 Protobuf 替代 Writable，提升序列化效率與跨語言兼容性