Hadoop系统通信协议介绍

转: http://weilaiyxj.iteye.com/blog/913166

本文约定：

DN: DataNode

TT: TaskTracker

NN: NameNode

SNN: Secondry NameNode

JT: JobTracker

本文介绍Hadoop各节点和Client之间通信协议。

Hadoop的通信是建立在RPC的基础上，关于RPC的详解介绍大家可以参照
"hadoop rpc机制 && 将avro引入hadoop rpc机制初探"

Hadoop中节点之间的通信是比较复杂的一个网络，若可以把它们之间的通信网络了解清楚，那么对于Hadoop的整体框架理解会有很大帮助。

HDFS中有5种协议：

DatanodeProtocol (DN && NN)

InterDatanodeProtocol (DN && DN)

ClientDatanodeProtocol (Client && DN)

ClientProtocol (Client && NN)

NamenodeProtocol (SNN && NN)

Map/Reduce 中有3种协议:

InterTrackerProtocol (TT && JT)

JobSubmissionProtocol (Client && JT)

TaskUmbilicalProtocol (Child && TT)

其中，DatanodeProtocol，ClientDatanodeProtocol，InterTrackerProtocol，TaskUmbilicalProtocol，JobSubmissionProtocol这5种协议通信频繁。

这8种协议在Hadoop中是作为接口存在，8种协议都继承接口VersionedProtocol。

协议的实现主要集中在类： JobTracker, NameNode, TaskTracker, DataNode 这4个类中。

其中

JobTracker实现两个接口： InterTrackerProtocol，JobSubmissionProtocol

TaskTracker类实现： TaskUmbilicalProtocol

NameNode类实现： DatanodeProtocol，ClientProtocol，NamenodeProtocol

DataNode类实现： InterDatanodeProtocol，ClientDatanodeProtocol

类图参照下图



下面单独介绍每种协议

1.InterTrackerProtocol (TaskTracker & JobTracker通信协议)

协议简介：

TT 每隔一段时间向JT发送heartbeat提交当前TT所在机器的内存，CPU，磁盘等状态信息，是否可以接收新任务等信息；JT接收到 heartbeat给TT返回需要所做的job或task操作，是否开启新的任务。TT可以从取得JT取得当前文件系统路径，需要执行Job的Jar文件路径等。

方法介绍:

heartbeat()    //TT周期性(最短时间缺省时间是3秒)的调用heartbeat给JT发送heartbeat

输入参数：

    TaskTrackerStatus: TaskTracker机器状态信息

空闲磁盘信息，虚拟和实际内存信息，Map使用内存，Reduce使用内存，可以虚拟和物理内存，累计 CPU 时间，CPU频率，CPU处理器个数，CPU使用率

    Restarted: 是否是第一次启动

    initialContact: 是否需要初始化

    acceptNewTasks: 是否能接收新的Task (Map/Reduce)

    responseId: 上次从JT成功接收的ID

返回结果：(JobTracker返回HeartbeatResponse )

  HeartbeatResponse对象有以下几个属性：

    responseId：返回应答ID

    heartbeatInterval: heartbeat间隔

    action列表: JT需要TT所做的操作(LAUNCH_TASK,KILL_TASK,KILL_JOB,  REINIT_TRACKER,COMMIT_TASK)

getFilesystemName: TaskTracker调用一次，取得的文件系统路径

reportTaskTrackerError: TT报告问题到JT

getTaskCompletionEvents: 取得jobid的task完成信息

getSystemDir: 取得执行Job所需要文件的路径

2. JobSubmissionProtocol(JobClient & JobTracker 之间通信)

协议简介：

主要用来提交，执行Job和取得当前Job状态；

JobClient可以通过此协议提交Job(submitJob),取得当前job的参数，状态，Counter，Task状态，文件系统路径，jar文件路径，所有job的状态，还有Job队列的信息，

此外JobClient还可以提交KillJob和设定Job优先级等任务到JobTracker；

主要方法：

getNewJobId：分配JobId

submitJob：提交Job执行，返回Job状态

getClusterStatus：取得当前Cluster的状态，如果detailed为true还报告tracker names信息，返回cluster状态

killJob：Kill Job

setJobPriority：设定job优先级

killTask：Kill 任务attempt

getJobProfile: 取得已运行的job的句柄

getJobStatus: 取得已运行Job的状态

getJobCounters: 取得job的Counter   

getMapTaskReports: 取得Job的各个Map Task报告

getReduceTaskReports: 取得Job的各个Reduce Task报告

getCleanupTaskReports：取得CleanupTask结果报告

getSetupTaskReports：取得Setup Task报告

getFilesystemName: 返回文件系统的名称

jobsToComplete: 取得没有完成和没有失败的Job状态(正在运行Job)

getAllJobs：取得所有Job状态

getTaskCompletionEvents： 取得特定Jobid完成的事件

getSystemDir：取得job文件存放的路径

getQueues：取得JobTracker的队列集合

getQueueInfo： 取得一个Job的调度信息

getJobsFromQueue：取得一个队列中所有提交的job 

3. TaskUmbilicalProtocol （Child & TaskTracker）

协议简介：

Child & TaskTracker之间通信协议：

Child通过此协议，查看TT是否存在，取得任务，报告任务的进度，状态，出错信息，Commit文件到HDFS，并取得map结果给reduce；TaskTracker接收任务并监控任务的进度

主要方法：

getTask:  Child运行时，调用此方法取得task任务

statusUpdate： 报告child进程进度

reportDiagnosticInfo：

reportNextRecordRange：报告Task下一步将要处理的记录范围 

ping： child周期性的调用此方法来查看parent(TaskTracker)是否存在
commitPending： 报告此任务任务已完成，但是文件还未提交(还未Commit)

canCommit： 轮询task是否可以提交文件(Commit)

getMapCompletionEvents：Reduce任务调用此方法取得Map输出文件, 返回map状态更新(map完成事务)

4. DatanodeProtocol (NN & DN)

协议简介：

DN注册信息到NN，发送当前DN,block等信息到NN(发送block报告，发送block错误报告)

NN返回DN需要执行的操作(删除block或copy)

主要方法：

register ：注册DN到NN

sendHeartbeat：  DN报告NN，存在以及当前状态; NN返回DN  DatanodeCommand数组，告诉DN需要做的操作(标记block失效，Copy等)

blockReport：  blockReport()告诉NameNode所有本地的块列表，NameNode返回一个废弃的需要删除的块的列表； 这个功能可以通知NameNode自己机器上的块的情况，当hadoop启动的时候，通过这个方法NameNode重建block和DN的关系表；其他情况会定期检查(缺省时间一个月)

blockReceived： DataNode告诉NameNode，最近收到的block (假如有超过重复因子数量的额外的块会删除) 例如：client写入一个新块，或另一个DN拷贝一个block到此DN，会调用此方法

errorReport: 报告NameNode 数据节点发生错误

processUpgradeCommand： 给NameNode发送升级命令，返回更新命令

reportBadBlocks： 报告错误的Block　

nextGenerationStamp： 返回某一块的新的时间戳
commitBlockSynchronization：
Commit block synchronization in lease recovery

5. InterDatanodeProtocol (DN & DN)

协议简介：

取得blockmeta，更新block信息等

主要方法：

getBlockMetaDataInfo： 取得某blockMeta信息

updateBlock：更新Block信息(主要是时间戳和长度)

6.NamenodeProtocol (NN & SNN)

协议简介：

取得DN块列表，回滚edit log, 回滚FsImage

主要方法：

getBlocks： 取得某DataNode的 BlockLIst 并且是固定的Size，返回一个block和位置数组

getEditLogSize：取得当前edit log的size

rollEditLog: 关闭旧的edit log，打开一个新edit log (返回CheckpointSignature ： 一个独立的事务ID)   

rollFsImage： 主要工作删除旧的fsImage, 拷贝新的image到fsImage; 删除旧的old edits, 重命名到edits

7.ClientDatanodeProtocol (Client && DN)

协议简介：

recoverBlock: 恢复block,返回new blockId和 generation stamp

8.ClientProtocol (Client & NN)

协议简介：

Client-->NN: 用户进行文件操作，文件系统操作，系统管理和问题处理的接口

主要方法：

getBlockLocations： 得到文件某个范围内块的DN列表( 排序是按照和client的远近来排序， 之后Client根据DN位置，直接去最近DN取得实际数据)，

create：  创建一个新的文件

append： 追加内容到一个文件结尾(dfs.support.append 为 true时 起作用)

setReplication： 设定当前文件的重复因子

setPermission： 设定存在的文件/目录的权限

setOwner: 设定一个路径的owner

abandonBlock: 丢弃一个block

addBlock: 文件里写入Block
complete：结束写数据，如果文件正常关闭true; 否则 false

reportBadBlocks： Client向NameNode报告损坏的Block

rename： 重命名一个文件或者目录

delete： 删除一个文件或者目录

mkdirs： 创建一个目录

getListing： 返回文件目录状态

renewLease： 判断client 是否dead

getStats： 取得fileSystem的状态

getDatanodeReport： 取得当前datanode的report

getPreferredBlockSize： 取得文件的block size

setSafeMode： 进入safe mode

saveNamespace:  保存namespace image

refreshNodes:  NameNode重新读取Hosts文件

finalizeUpgrade： 结束上次更新

metaSave： 将当前系统namenode数据结构dump到一个文件，如果存在那么追加

getFileInfo： 取得文件的信息

setQuota： 设定文件目录的配额

getContentSummary： 目录的内容摘要

fsync: 同步文件所有metadata到文件

setTimes： 设定文件时间

注：本文写作基于hadoop 0.20.2，其他版本某些函数会有改变

想共同探讨Hadoop请联系作者： hadoop123 at gmail.com