hadoop的集群是基于master/slave模式,namenode和jobtracker属于master,而datanode/tasktracker属于slaves。master只有一个,而slaves有多个。
namenode与datanode之间的通信,jobtracker与tasktracker直接的通信,都是通过“心跳”完成的。
以前看过hadoop心跳原理的源代码,今天再回忆一下,呵呵,所以叫“心跳回忆”。
1、心跳机制
心跳的机制大概是这样的:
1) master启动的时候,会开一个ipc server在那里。
2) slave启动时,会连接master,并每隔3秒钟主动向master发送一个“心跳”,将自己的状态信息告诉master,然后master也是通过这个心跳的返回值,向slave节点传达指令。
2、找到心跳的代码
拿namenode和datanode来说,在datanode的offerService方法中,每隔3秒向namenode发送心跳的代码:
/**
* Main loop for the DataNode. Runs until shutdown,
* forever calling remote NameNode functions.
*/
public void offerService() throws Exception {
...
//
// Now loop for a long time....
//
while (shouldRun) {
try {
long startTime = now();
//
// Every so often, send heartbeat or block-report
//
// 如果到了3秒钟,就向namenode发心跳
if (startTime - lastHeartbeat > heartBeatInterval) {
//
// All heartbeat messages include following info:
// -- Datanode name
// -- data transfer port
// -- Total capacity
// -- Bytes remaining
//
lastHeartbeat = startTime;
DatanodeCommand[] cmds = namenode.sendHeartbeat(dnRegistration,
data.getCapacity(),
data.getDfsUsed(),
data.getRemaining(),
xmitsInProgress.get(),
getXceiverCount());
// 注意上面这行代码,“发送心跳”竟然就是调用namenode的一个方法??
myMetrics.heartbeats.inc(now() - startTime);
//LOG.info("Just sent heartbeat, with name " + localName);
// 处理对心跳的返回值(namenode传给datanode的指令)
if (!processCommand(cmds))
continue;
}
// 这里省略很多代码
...
} // while (shouldRun)
} // offerService
上面这段代码,如果是单机的程序,没什么值得奇怪的。但是,这是hadoop集群!datanode和namenode在2台不同的机器(或2个JVM)上运行!datanode机器竟然直接调用namenode的方法!这是怎么实现的?难道是传说中的RMI吗??
下面我们主要就来分析这个方法调用的细节。
3、心跳的底层细节一:datanode怎么获得namenode对象的?
首先,DataNode类中,有一个namenode的成员变量:
public class DataNode extends Configured
implements InterDatanodeProtocol, ClientDatanodeProtocol, FSConstants, Runnable {
...
public DatanodeProtocol namenode = null;
...
}
下面是NameNode类的定义:
public class NameNode implements ClientProtocol, DatanodeProtocol,
NamenodeProtocol, FSConstants,
RefreshAuthorizationPolicyProtocol {
...
}
注意:NameNode实现了DatanodeProtocol接口,DatanodeProtocol接口定义了namenode和datanode之间通信的方法。
那么,DataNode类是怎么获取到NameNode类的引用呢?
在Datanode端,为namenode变量赋值的代码:
// connect to name node
this.namenode = (DatanodeProtocol)
RPC.waitForProxy(DatanodeProtocol.class,
DatanodeProtocol.versionID,
nameNodeAddr,
conf);
在继续去RPC类中追踪:
VersionedProtocol proxy =
(VersionedProtocol) Proxy.newProxyInstance(
protocol.getClassLoader(), new Class[] { protocol },
new Invoker(addr, ticket, conf, factory));
现在,明白了!
1) 对namenode的赋值,并不是真正的new了一个实现了DatanodeProtocol接口的对象,而是获得了一个
动态代理!!
2) 上面这段代码中,protocol的类型是DatanodeProtocol.class
3) 对namenode的所有调用,都被委托(delegate)给了Invoker
4、心跳的底层细节二:看看Invoker类
Invoker类是org.apache.hadoop.ipc.RPC类的一个静态内部类:
private static class Invoker implements InvocationHandler {
在这个类中,看invoke方法:
public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
...
ObjectWritable value = (ObjectWritable)
client.call(new Invocation(method, args), address,
method.getDeclaringClass(), ticket);
...
return value.get();
}
所有的方法调用又被delegate给client的call方法了!
client是Invoker中的成员变量:
private Client client;
所以可以看出:DatanodeProtocol中的每个方法调用,都被包装成一个Invocation对象,再由client.call()调用
5、心跳的底层细节三:Invocation类
Invocation类是org.apache.hadoop.ipc.RPC类的一个静态内部类
没有什么业务逻辑方法,主要作用就是一个VO
6、心跳的底层细节四:client类的call方法
接下来重点看client类的call方法:
public Writable call(Writable param, InetSocketAddress addr,
Class<?> protocol, UserGroupInformation ticket)
throws InterruptedException, IOException {
Call call = new Call(param);
// 将Invocation转化为Call
Connection connection = getConnection(addr, protocol, ticket, call);
// 连接远程服务器
connection.sendParam(call); // send the parameter
// 将“序列化”后的call发给过去
boolean interrupted = false;
synchronized (call) {
while (!call.done) {
try {
call.wait(); // wait for the result
// 等待调用结果
} catch (InterruptedException ie) {
// save the fact that we were interrupted
interrupted = true;
}
}
if (interrupted) {
// set the interrupt flag now that we are done waiting
Thread.currentThread().interrupt();
}
if (call.error != null) {
if (call.error instanceof RemoteException) {
call.error.fillInStackTrace();
throw call.error;
} else { // local exception
throw wrapException(addr, call.error);
}
} else {
return call.value;
// 返回
}
}
}
7、现在,一目了然了
datanode向namenode发送heartbeat过程是这样的:
a) 在datanode初始化获得namenode的proxy
b) 在datanode上,调用namenode proxy的heartbeat方法:
namenode.sendHeartbeat(dnRegistration,
data.getCapacity(),
data.getDfsUsed(),
data.getRemaining(),
xmitsInProgress.get(),
getXceiverCount());
c) 在datanode上的namenode动态代理类将这个调用包装成(或者叫“序列化成”)一个Invocation对象,并调用client.call方法
d) client call方法将Invocation转化为Call对象
e) client 将call发送到真正的namenode服务器
f) namenode接收后,转化成namenode端的Call,并process后,通过Responder发回来!
g) datanode接收结果,并将结果转化为DatanodeCommand[]
8、再看动态代理
动态代理:让“只有接口,没事对应的实现类”成为可能,因为具体方法的实现可以委托给另一个类!!
在这个例子中,就datanode而言,DatanodeProtocol接口是没有实现类的!
*** THE END ***
分享到:
相关推荐
在windows环境下开发hadoop时,需要配置HADOOP_HOME环境变量,变量值D:\hadoop-common-2.7.3-bin-master,并在Path追加%HADOOP_HOME%\bin,有可能出现如下错误: org.apache.hadoop.io.nativeio.NativeIO$Windows....
《Hadoop大数据开发实战》教学教案—01初识Hadoop.pdf《Hadoop大数据开发实战》教学教案—01初识Hadoop.pdf《Hadoop大数据开发实战》教学教案—01初识Hadoop.pdf《Hadoop大数据开发实战》教学教案—01初识Hadoop.pdf...
Hadoop 是一个处理、存储和分析海量的分布式、非结构化数据的开源框架。最初由 Yahoo 的工程师 Doug Cutting 和 Mike Cafarella Hadoop 是一个处理、存储和分析海量的分布式、非结构化数据的开源框架。最初由 Yahoo...
Hadoop是一个由Apache基金会所开发的分布式系统基础架构。用户可以在不了解分布式底层细节的情况下,开发分布式程序。充分利用集群的威力进 Hadoop是一个由Apache基金会所开发的分布式系统基础架构。用户可以在不...
本书从hadoop的缘起开始,由浅入深,结合理论和实践,全方位地介绍hado叩这一高性能处理海量数据集的理想工具。全书共14章,3个附录,涉及的主题包括:haddoop简介:mapreduce简介:hadoop分布式文件系统;hadoop的i...
hadoop_tutorial hadoop入门经典 Hadoop 是一个能够对大量数据进行分布式处理的软件框架。Hadoop 是可靠的,因为它假设计算元素和存储会失败,因此它维护多个工作数据副本,确保能够针对失败的节点重新分布处理。...
hadoop2.7.3 Winutils.exe hadoop.dll
hadoop的dll文件 hadoop.zip
hadoop-annotations-3.1.1.jar hadoop-common-3.1.1.jar hadoop-mapreduce-client-core-3.1.1.jar hadoop-yarn-api-3.1.1.jar hadoop-auth-3.1.1.jar hadoop-hdfs-3.1.1.jar hadoop-mapreduce-client-hs-3.1.1.jar ...
Hadoop 集群配置详解 Hadoop_Hadoop集群(第1期)_CentOS安装配置 Hadoop_Hadoop集群(第2期)_机器信息分布表 Hadoop_Hadoop集群(第4期)_SecureCRT使用 Hadoop_Hadoop集群(第5期)_Hadoop安装配置 Hadoop_Hadoop...
调用保存文件的算子,需要配置Hadoop依赖 将文件夹中的 hadoop-3.0.0 解压到电脑任意位置 在Python代码中使用os模块配置:os.environ[‘HADOOP_HOME’] = ‘HADOOP解压文件夹路径’ winutils.exe,并放入Hadoop解压...
Apache Hadoop (hadoop-3.3.4.tar.gz)项目为可靠、可扩展的分布式计算开发开源软件。官网下载速度非常缓慢,因此将hadoop-3.3.4 版本放在这里,欢迎大家来下载使用! Hadoop 架构是一个开源的、基于 Java 的编程...
Hadoop大数据资料集锦Hadoop大数据资料集锦Hadoop大数据资料集锦Hadoop大数据资料集锦
hadoop2.6.0 hadoop.dll包括winutils.exe
Hadoop实现了一个分布式文件系统(Hadoop Distributed File System),简称HDFS。HDFS有高容错性的特点,并且设计用来部署在低廉的(low-cost)硬件上;而且它提供高吞吐量(high throughput)来访问应用程序的数据,适合...
支持如下版本的Hadoop hadoop-2.6.0 hadoop-2.6.3 hadoop-2.6.4 hadoop-2.7.1 hadoop-2.8.1 hadoop-2.8.3 hadoop-3.0.0
hadoop hadoop的hadoop.dll和winutils.exe 解决方法, 把winutils.exe加入你的hadoop-x.x.x/bin下 Could not locate executable null\bin\winutils.exe in the Hadoop binaries
hadoop的hadoop.dll和winutils.exe下载
windows下做hadoop入门,会出现hdfs报错,2.7.7版本兼容 windows下做hadoop入门,会出现hdfs报错,2.7.7版本兼容 windows下做hadoop入门,会出现hdfs报错,2.7.7版本兼容
Hadoop豆瓣电影数据分析(Hadoop)操作源码