Hadoop源码分析之DataNode的启动与停止

时间 2014-02-17 17:12:50 CSDN博客

原文 http://blog.csdn.net/workformywork/article/details/18792637
主题 Hadoop

DataNode的启动分可以为两个部分，创建DataNode对象和启动DataNode节点。DataNode.main()方法会调用DataNode.secureMain()方法，接着在secureMain方法中调用DataNode.createDataNode()方法，createDataNode方法调用了DataNode类中的instantiateDataNode方法和runDatanodeDaemon方法，其中instantiateDataNode方法用于初始化DataNode的大部分成员变量，即创建DataNode对象，runDatanodeDaemon方法用于向NameNode节点注册和启动DataNode节点的线程，即启动DataNode线程。

创建DataNode对象

创建DataNode对象调用了DataNode类的静态方法instantiateDataNode，方法的代码如下：

public static DataNode instantiateDataNode(String args[],
Configuration conf,
SecureResources resources) throws IOException {
if (conf == null)
conf = new Configuration();//创建Configuration对象
if (!parseArguments(args, conf)) {
printUsage();
System.exit(-2);
}
if (conf.get("dfs.network.script") != null) {
LOG.error("This configuration for rack identification is not supported" +
" anymore. RackID resolution is handled by the NameNode.");
System.exit(-1);
}
String[] dataDirs = conf.getStrings(DATA_DIR_KEY);//获取数据块的存储目录
dnThreadName = "DataNode: [" +
StringUtils.arrayToString(dataDirs) + "]";
DefaultMetricsSystem.initialize("DataNode");
return makeInstance(dataDirs, conf, resources);//创建DataNode对象
}

在这个方法中，先创建一个Configuration对象，然后解析启动参数，再获取到所配置的存储数据块的目录，这个属性由${dfs.data.dir}这个属性指明，如果在自定义的xml文件中没有指定，则会读取hdfs-default.xml文件，那么会读取到${hadoop.tmp.dir}/dfs/data 这个值，即${hadoop.tmp.dir}目录下的dfs/data目录。方法的最后调用makeInstance方法创建DataNode对象，传入的参数分别是数据块的存储目录，配置信息，所需要的资源（执行DataNode.main()方法时，这个参数为null）。

DataNode.makeInstance()方法中，先检查${dfs.data.dir}所指定的目录，当至少有一个目录存在且其所属用户的有可读和可写的权限时才调用DataNode的构造方法创建DataNode对象，方法的代码如下：

public static DataNode makeInstance(String[] dataDirs, Configuration conf,
SecureResources resources) throws IOException {
UserGroupInformation.setConfiguration(conf);
LocalFileSystem localFS = FileSystem.getLocal(conf);//获取本地文件系统
ArrayList<File> dirs = new ArrayList<File>();
FsPermission dataDirPermission =
new FsPermission(conf.get(DATA_DIR_PERMISSION_KEY,
DEFAULT_DATA_DIR_PERMISSION));
for (String dir : dataDirs) {
try {
DiskChecker.checkDir(localFS, new Path(dir), dataDirPermission);//检查文件夹在本地文件系统中是否存在，如果存在就检查文件所属用户的权限
dirs.add(new File(dir));//将这个目录加入到列表中
} catch(IOException e) {
LOG.warn("Invalid directory in " + DATA_DIR_KEY +  ": " +
e.getMessage());
}
}
if (dirs.size() > 0) //至少有一个目录存在，且权限满足
return new DataNode(conf, dirs, resources);
LOG.error("All directories in " + DATA_DIR_KEY + " are invalid.");
return null;
}

makeInstance方法调用了DiskChecker类的静态方法checkDir方法来检查DataNode节点是否存在指定的文件夹，如果不存在则会创建，并设置权限，再检查文件夹的权限是否满足读写要求，如果执行checkDir方法的过程中出错，则会抛出IOException异常，正常执行则会将这个文件夹对应的文件对象加入到列表dirs中。最后调用DataNode的构造方法来创建DataNode对象。

DataNode的构造方法代码如下：

DataNode(final Configuration conf,
final AbstractList<File> dataDirs, SecureResources resources) throws IOException {
super(conf);
SecurityUtil.login(conf, DFSConfigKeys.DFS_DATANODE_KEYTAB_FILE_KEY,
DFSConfigKeys.DFS_DATANODE_USER_NAME_KEY);

datanodeObject = this;
durableSync = conf.getBoolean("dfs.durable.sync", true);
this.userWithLocalPathAccess = conf
.get(DFSConfigKeys.DFS_BLOCK_LOCAL_PATH_ACCESS_USER_KEY);
try {
startDataNode(conf, dataDirs, resources);
} catch (IOException ie) {
shutdown();
throw ie;
}
}

在此方法中，先设置Configuration对象（DataNode类继承自Configured类，有一个Configuration类型的conf变量）。方法中调用的SecurityUtil.login()方法只有在配置了Kerberos认证时才会起作用，如果没有配置则会在方法中直接返回。然后就是为几个成员变量赋值，再调用startDataNode()方法完成DataNode类的初始化。DataNode.startDataNode()方法的代码比较长，这里分几个部分进行分析。

InetSocketAddress nameNodeAddr = NameNode.getServiceAddress(conf, true);//获取NameNode的地址

这行代码是获取NameNode节点的地址，DataNode节点使用这个地址与NameNode节点进行IPC通信。在NameNode节点中，存在这两个进行IPC通信的地址，关于这两个进行IPC通信的地址的内容可以参见文章 Hadoop源码分析之NameNode的启动与停止 ，此处通过NameNode.getServiceAddress()方法获取与服务器进行通信的地址。

//默认当NameNode与DataNode的build版本不一致时，DataNode拒绝连接到NameNode，这个选项如果为true，则只检查版本，而不检查其他
this.relaxedVersionCheck = conf.getBoolean(
CommonConfigurationKeys.HADOOP_RELAXED_VERSION_CHECK_KEY,
CommonConfigurationKeys.HADOOP_RELAXED_VERSION_CHECK_DEFAULT);
//默认当NameNode与DataNode的build版本不一致时，DataNode拒绝连接到NameNode，这个选项如果为true，则不检查版本，并且这个选项值可覆盖relaxedVersionCheck
noVersionCheck = conf.getBoolean(
CommonConfigurationKeys.HADOOP_SKIP_VERSION_CHECK_KEY,
CommonConfigurationKeys.HADOOP_SKIP_VERSION_CHECK_DEFAULT);

这两行代码主要用于检查NameNode节点和DataNode节点的版本是否一致，Hadoop中默认情况下NameNode与DataNode版本不一致时，DataNode会拒绝连接到NameNode，所以需要通过一定的方式来检查NameNode和DataNode的版本是否一致。如果hadoop.relaxed.worker.version.check属性设置为true，则只会检查版本号，而不会去检查其他的字段，而如果属性hadoop.skip.worker.version.check设置为true时，就会跳过版本号的检查，即什么都不检查。

InetSocketAddress socAddr = DataNode.getStreamingAddr(conf);//数据节点提供数据流服务的地址
int tmpPort = socAddr.getPort();
storage = new DataStorage();
// construct registration
this.dnRegistration = new DatanodeRegistration(machineName + ":" + tmpPort);//数据节点在NameNode节点注册的对象

这段代码首先获取管理员所配置的数据节点提供数据流服务的网络地址(IP+端口)，然后创建一个DatanodeRegistration对象，用于数据节点向NameNode进行注册。DataNode.getStreamingAddr()方法会读取配置中的${dfs.datanode.bindAddress}，${dfs.datanode.port}和${dfs.datanode.address}，，其中前两个是老版本的Hadoop配置方式，分IP和端口配置，后一个是新版本中的配置方式，如果${dfs.datanode.address}配置不为空则，直接返回，否则使用${dfs.datanode.bindAddress}和${dfs.datanode.port}的值。

// connect to name node，调用RPC.waitForProxy()方法得到一个动态DatanodeProtocol对象的引用，在waitForProxy方法中调用了getProxy方法，IPC中分析过
//getProxy方法
this.namenode = (DatanodeProtocol)RPC.waitForProxy(DatanodeProtocol.class,DatanodeProtocol.versionID,nameNodeAddr, conf);

这行代码通过waitForProxy方法创建一个DatanodeProtocol类型的引用，使用分析IPC时分析到的getProxy方法中的动态代理方式创建，这个方法的实际调用代码如下：

static VersionedProtocol waitForProxy(
Class<? extends VersionedProtocol> protocol,
long clientVersion,
InetSocketAddress addr,
Configuration conf,
int rpcTimeout,
long connTimeout)
throws IOException {
long startTime = System.currentTimeMillis();
IOException ioe;
while (true) {
try {
return getProxy(protocol, clientVersion, addr, conf, rpcTimeout);
} catch(ConnectException se) {  // namenode has not been started
LOG.info("Server at " + addr + " not available yet, Zzzzz...");
ioe = se;
} catch(SocketTimeoutException te) {  // namenode is busy
LOG.info("Problem connecting to server: " + addr);
ioe = te;
}
// check if timed out
if (System.currentTimeMillis()-connTimeout >= startTime) {
throw ioe;
}

// wait for retry
try {
Thread.sleep(1000);
} catch (InterruptedException ie) {
// IGNORE
}
}
}

由代码可以看出在一个while循环中调用getProxy()方法，直到调用成功时才返回，否则就一直停留在while循环中，每隔1分钟调用一次getProxy方法，之所以要循环调用getProxy()方法，是因为DataNode与NameNode进行网络通信时可能会出现各种各样的问题致使连接不到，所以将该方法放在一个循环中，可以避免连接不到的问题。

NamespaceInfo nsInfo = handshake();

这行代码调用NameNode.handshake()方法，从NameNode节点中获取版本号和ID信息，方法的代码如下：

/**
* 与NameNode握手，数据节点会保证它的构建信息和存储系统结构版本号和名字节点的一致
* @return
* @throws IOException
*/
private NamespaceInfo handshake() throws IOException {
NamespaceInfo nsInfo = new NamespaceInfo();
while (shouldRun) {
try {
nsInfo = namenode.versionRequest();//调用远程方法，获取名字节点的信息
break;
} catch(SocketTimeoutException e) {  // namenode is busy
LOG.info("Problem connecting to server: " + getNameNodeAddr());
try {
Thread.sleep(1000);
} catch (InterruptedException ie) {}
}
}
if (!isPermittedVersion(nsInfo)) {
String errorMsg = "Shutting down. Incompatible version or revision." +
"DataNode version '" + VersionInfo.getVersion() +
"' and revision '" + VersionInfo.getRevision() +
"' and NameNode version '" + nsInfo.getVersion() +
"' and revision '" + nsInfo.getRevision() +
" and " + CommonConfigurationKeys.HADOOP_RELAXED_VERSION_CHECK_KEY +
" is " + (relaxedVersionCheck ? "enabled" : "not enabled") +
" and " + CommonConfigurationKeys.HADOOP_SKIP_VERSION_CHECK_KEY +
" is " + (noVersionCheck ? "enabled" : "not enabled");
LOG.fatal(errorMsg);
notifyNamenode(DatanodeProtocol.NOTIFY, errorMsg);
throw new IOException( errorMsg );
}
assert FSConstants.LAYOUT_VERSION == nsInfo.getLayoutVersion() :
"Data-node and name-node layout versions must be the same."
+ "Expected: "+ FSConstants.LAYOUT_VERSION + " actual "+ nsInfo.getLayoutVersion();
return nsInfo;
}

这个方法先调用DatanodeProtocol动态代理对象namenode的versionRequest()方法，向NameNode发送一次远程方法调用，用于获取名字节点信息，这次远程方法调用是在一个循环中进行的，直到调用成功为止。远程方法调用成功则返回一个NamespaceInfo对象，再调用DataNode.isPermittedVersion()方法来检查，DataNode的版本是否与NameNode的版本一致。在DataNode.isPermittedVersion()方法中使用到了两个变量进行控制，分别是noVersionCheck和relaxedVersionCheck，上面已经介绍过了，不再重复。如果版本检查不通过，则会抛出异常。

boolean simulatedFSDataset = conf.getBoolean("dfs.datanode.simulateddatastorage", false);

这行代码获取属性dfs.datanode.simulateddatastorage的值，默认是false，这个属性模拟集群环境，用在开发调试时，在一台机器上模拟分布式环境，参见https://issues.apache.org/jira/browse/HADOOP-1989 。

如果是没有配置这个属性，则会执行下面的代码：

//先将存储目录从某一状态中恢复，再读取目录中的信息
storage.recoverTransitionRead(nsInfo, dataDirs, startOpt);
// adjust
this.dnRegistration.setStorageInfo(storage);
// initialize data node internal structure
this.data = new FSDataset(storage, conf);

其中Storage.recoverTransitionRead()方法与FSImage.recoverTransitionRead()方法类似，先分析存储目录的状态，如果存储目录处于一个临时状态则从将目录上一次运行状态中恢复，然后根据nsInfo执行执行状态转换，最后读入目录中存储的信息到内存。

接下来是数据节点间交换数据的流式接口服务器的创建，数据节点流式接口服务器使用了Socket来进行节点间的通信，在startDataNode()方法中，相关代码为：

// find free port or use privileged port provide
ServerSocket ss;
if(secureResources == null) {
ss = (socketWriteTimeout > 0) ?
ServerSocketChannel.open().socket() : new ServerSocket();
Server.bind(ss, socAddr, 0);
} else {
ss = resources.getStreamingSocket();
}
//默认为8KB，这里设置为128KB，因为数据节点需要提供搞吞吐率的数据服务，所以需要较大的缓冲区，这个缓冲区适用于所有从accept方法返回的参数
ss.setReceiveBufferSize(DEFAULT_DATA_SOCKET_SIZE);
this.dnRegistration.setName(machineName + ":" + tmpPort);

其中socAddr就是上面分析到的sockAddr对象，首先创建一个ServerSocket对象，然后为这个ServerSocket对象绑定一个IP，再设置接收缓存大小，ServerSocket默认的接收缓存大小为8KB，这里设置为128KB，因为数据节点需要提供搞吞吐率的数据服务，所以需要较大的缓冲区，这个缓冲区适用于所有从accept方法返回的参数。此外，执行ServerSocket.bind()方法是可能绑定的端口不是给定的端口，所以带有端口号的用于数据节点注册的对象dnRegistration对象要重新设置一次name变量。

//使用一个线程组
this.threadGroup = new ThreadGroup("dataXceiverServer");
this.dataXceiverServer = new Daemon(threadGroup,
new DataXceiverServer(ss, conf, this));//DataXceiverServer中有一个ServerSocket对象，用于接收客户端Socket
this.threadGroup.setDaemon(true); // auto destroy when empty，设置线程组中所有的线程为守护线程

这段代码为数据节点流式接口服务线程建立线程组，创建DataXceiverServer服务器，用于数据节点间交换数据流，并且将线程组中的线程设置为守护线程，这样就可以对提供流式服务的线程进行统一的操作，由于线程组中的线程又是守护线程，方便了对各个线程的管理。

InetSocketAddress ipcAddr = NetUtils.createSocketAddr(
conf.get("dfs.datanode.ipc.address"));
ipcServer = RPC.getServer(this, ipcAddr.getHostName(), ipcAddr.getPort(),
conf.getInt("dfs.datanode.handler.count", 3), false, conf,
blockTokenSecretManager);
dnRegistration.setIpcPort(ipcServer.getListenerAddress().getPort());

这段代码创建一个数据节点之间进行IPC通信的对象ipcServer。

DataNode类中使用到的变量infoServer是在DataNode节点中建立内置的HTTP服务器，用于方便在浏览器中查看DataNode的状态信息。

启动DataNode节点

创建了DataNode对象之后，就可以向NameNode节点注册，然后启动DataNode线程了。在DataNode.createDataNode()方法中，上述的创建DataNode对象的过程是执行DataNode的静态方法DataNode.instantiateDataNode()，启动DataNode节点则是执行了静态方法DataNode.runDatanodeDaemon()，该方法的代码如下：

/** Start a single datanode daemon and wait for it to finish.
*  If this thread is specifically interrupted, it will stop waiting.
*/
public static void runDatanodeDaemon(DataNode dn) throws IOException {
if (dn != null) {
//register datanode
dn.register();
dn.dataNodeThread = new Thread(dn, dnThreadName);
dn.dataNodeThread.setDaemon(true); // needed for JUnit testing
dn.dataNodeThread.start();
}
}

在方法中，首先调用DataNode.register()向NameNode节点注册，register()方法进行必要的数据节点初始化的工作，然后调用远程方法向NameNode节点进行注册，再随机等待一定时间后上报数据节点上正处于写状态的数据块，具体代码如下：

private void register() throws IOException {
if (dnRegistration.getStorageID().equals("")) {
setNewStorageID(dnRegistration);//构造一个新的数据节点标识
}
while(shouldRun) {
try {
// reset name to machineName. Mainly for web interface.
dnRegistration.name = machineName + ":" + dnRegistration.getPort();//注册信息
dnRegistration = namenode.register(dnRegistration);
break;
} catch(SocketTimeoutException e) {  // namenode is busy
try {
Thread.sleep(1000);
} catch (InterruptedException ie) {}
}
}
//数据节点注册后还需要根据目前数据节点的配置情况进执行一些后续处理
if (storage.getStorageID().equals("")) {
storage.setStorageID(dnRegistration.getStorageID());
storage.writeAll();//完成数据节点的存储初始化
}
if(! storage.getStorageID().equals(dnRegistration.getStorageID())) {
throw new IOException("Inconsistent storage IDs. Name-node returned "
+ dnRegistration.getStorageID()
+ ". Expecting " + storage.getStorageID());
}
if (durableSync) {
Block[] bbwReport = data.getBlocksBeingWrittenReport();//获取blocksBeingWritten目录下面的所有Block封装成Block[]数组
long[] blocksBeingWritten =
BlockListAsLongs.convertToArrayLongs(bbwReport);
namenode.blocksBeingWrittenReport(dnRegistration, blocksBeingWritten);
}

// random short delay - helps scatter the BR from all DNs
// - but we can start generating the block report immediately
data.requestAsyncBlockReport();
//在随机等待一小段时间后，DataNode的主循环offerService()方法就会使用blockReport()上报数据节点上现有的数据块信息
scheduleBlockReport(initialBlockReportDelay);
}

方法首先判断storageID是否已经生成，如果没有生成就调用DataNode.setNewStorageID()方法为当前的DataNode节点生成一个storageID。storageID是该数据节点在集群中的唯一标识，所以必须保证生成的这个字符串在整个HDFS集群中唯一，setNewStorageID()方法采取的方式是：DS+随机数+数据节点IP地址+提供流式数据服务的端口号+系统当前时间，这样就几乎可以产生不重复的值，具体代码参见DataNode.setNewStorageID()方法。然后，就向NameNode节点注册，可以看到，向NameNode节点注册的代码是在一个循环中执行，如果执行不成功就一直重复，直到注册成功。DataNode节点注册成功之后，数据节点对应的VERSION文件中的信息都生成了，此时就将这些信息写入到VERSION文件，代码storage.writeAll()来完成此操作，然后就根据变量durableSync看是否需要上报正处于写状态的数据块，durableSync变量由属性 dfs.durable.sync 指定，如果该变量为true，则查看数据节点中blockBeingWritten目录，读取该目录中的内容，然后上报给NameNode节点。scheduleBlockReport()方法根据变量initialBlockReportDelay变量的值来判断是下次是向NameNode节点发送心跳还是上报数据块。

DataNode.register()方法执行完成后，在DataNode.runDatanodeDaemon()方法中就创建DataNode线程（DataNode实现了Runnable接口），并设置该线程为守护线程，再启动该线程，然后就执行DataNode.run()方法了，该方法的代码如下：

public void run() {
// start dataXceiveServer
dataXceiverServer.start();
ipcServer.start();

while (shouldRun) {
try {
startDistributedUpgradeIfNeeded();
offerService();
} catch (Exception ex) {
LOG.error("Exception: " + StringUtils.stringifyException(ex));
if (shouldRun) {
try {
Thread.sleep(5000);
} catch (InterruptedException ie) {
}
}
}
}

在该线程中，首先启动了DataNode提供流式数据交换的服务和在各个数据节点间进行IPC通信的服务，然后进入到一个循环，调用startDistributedUpgradeIfNeeded()方法和offerService()方法，offerService()用于数据节点定期向NameNode节点上报数据块或者发送心跳。

这样就完成了数据节点的初始化与启动，以后再对数据节点中一些方法进行更深入的分析。

Reference

《Hadoop技术内幕：深入理解Hadoop Common和HDFS架构设计与实现原理》

/article/11519686.html