hadoop源代码分析（1）-hdfs.server.datanode包-DataNode类【原创】

dog1888

一  准备
　　　　hadop版本：1.0.3，DataNode所在的包：org.apache.hadoop.hdfs.server.datanode
　　　　学习方法：整理datanode类重要的方法、属性并理解，参考相关博客分析，最终理解datanode功能，再深入研究具体代码。
　　　　时间：2013-01-22  -- 2013-01-28


二  DataNode功能描述
　　　　DataNode是一个为分布式文件系统的调用存储数据块集的类。简单的调用就能启动一个或很多个数据节点。每个数据节点通常和一单个主节点通信，同时它也随时和客户端、其他数据节点保持通信。
　　　　数据节点存储着一系列数据块，它允许客户端去读这些数据块或者写入新的数据块。数据节点也可能收到namenode的指令，删除数据块或从其他datanodes拷贝数据块。
　　　　datanode维护着一张临界表：block->字节流
　　　　这个信息存储在本地磁盘上，数据节点在启动和每隔一段时间把这张表的内容报告给主节点。
　　　　Datanodes在它们的生命周期内维持着一个请求NameNode安排事情让他们去做。主节点不能直接和datanode连接，它只简单地返回数据节点的函数请求值。
　　　　数据节点维护着一个开放的服务socket，因此client或其他datanodes能够读/写数据。这个主机/端口会被报告给主节点，然后主节点把信息发送给感兴趣的客户端或其他数据节点。
　　　　从上面的描述我们可以看出datanode完成的功能很复杂，归纳来说大致有一下功能：启动、运行、通信、存储、安全保证。

三  DataNode如何实现其功能

1、DataNode启动
　　　　类开始，读取配置文件。
　　


View Code 　　
　　
　　　　由DataNode源代码可以看出，当DataNode启动时，依次执行以下方法：
　　a、main(String args[]) :void // 主入口
　　
　　　　执行b，安全的创建启动线程（datanode）方法。
　　b、secureMain(String[] args,SecureResources resources): void,参数args和null。
　　　　执行C，
　　　　记录DataNode启动和关闭的日志；
　　　　创建DataNode。
　　　　如果datanode不是空引用那么等待datanode线程终止。
　　c、StringUtils.startupShutdownMessage(DataNode.class, args, LOG);
      createDataNode(String args[],Configuration conf, SecureResources resources)：DataNode，参数：args, null, resources。调用instantiateDataNode(args, conf, resources);初始化DataNode，然后让数据节点在后台运行。
　　d、instantiateDataNode(args, conf, resources)：DataNode，首先配置文件如果为空，采用默认配置，解析命令行传入参数，启动选项只允许传入-rollback和-regular如果不是，返回空；如果设置类机架配置参数（"dfs.network.script"），程序退出。在这个函数里调用makeInstance(dataDirs, conf, resources)实例化DataNode。
　　e、makeInstance(String[] dataDirs, Configuration conf, SecureResources resources)：DataNode

• 通过conf，设置用户组信息
  
• 通过conf，获取本地文件系统
  
• 通过conf，获取文件系统的权限问题，通过org.apache.hadoop.fs.permission包提供
  
• 通过DiskChecker检查目录，首先检查存在并创建目录，然后验证权限。
  
• 通过DataNode构造方法，创建DataNode实例。
  

　　f、  DataNode(final Configuration conf, final AbstractList dataDirs, SecureResources resources)
　　构造方法。

• 使用安全工具类的SecurityUtil.login(conf, FSConfigKeys.DFS_DATANODE_KEYTAB_FILE_KEY,
  

　　        DFSConfigKeys.DFS_DATANODE_USER_NAME_KEY)方法登录，如果提供类keytab文件，那么用那个用户登录。用当前主机动态寻找完全符合的域名用户的Kerberos替代$host。用户所在文件的配置参数为“dfs.datanode.keytab.file”，用户名的配置参数为"dfs.datanode.kerberos.principal"，配置可参考《hadoop 添加kerberos认证》这篇文章。这也是以往在配置hadoop的时候一直不知道如何制定hadoop的权限问题，这回大致知道了解了一些。

• 获取当前DataNode的引用，datanodeObject；
  
• 获取supportAppends。
  
• 启动datanode。方法：startDataNode(conf,dataDirs,resouces)；
  

　　g、startDataNode(Configuration conf, AbstractList dataDirs, SecureResources resources)：void

• 如果用户组信息登录权限为真，安全资源为空，抛出运行异常（不能在没有权限的情况下启动集群）
  
• 通过conf.get("slave.host.name")，获取主机名；
  
• 执行NameNode.getServiceAddress(conf, true)，获取主节点InetSocketAddress地址,获取配置文件中fs.default.name指定的IP和端口；
  
• 通过conf获取socket超时时间，超时写入时间，写入包的大小
  
• 执行DataNode.getStreamingAddr(conf)，获取数据节点InetSocketAddress地址，获取端口号。
  
• 初始化DATASTORAGE。
  
• 注册数据节点，格式为主机名：端口号
  
• 连接到namenode，采用方法：RPC.waitForProxy(DatanodeProtocol.class, DatanodeProtocol.versionID, nameNodeAddr, conf);通过Hadoop的RPC机制与NameNode进行连接，获取namenode变量，这是DataNode与NameNode进行交互的工具,这个方法我没理解namenode怎么去判断是哪个datanode去连接的它，通过conf吗？
  
• 获取版本和id；
  
• 从配置文件读取DataNode启动选项，获取DataNode启动的模式是regular还是format，如果是第一次启动必须指定format，不指定默认是regular。有以下这几种选项
  
  

　　　　　　FORMAT  ("-format"),
    　　　REGULAR ("-regular"),
　　　　UPGRADE ("-upgrade"),
    　　　ROLLBACK("-rollback"),
    　　　FINALIZE("-finalize"),
    　　　IMPORT  ("-importCheckpoint");
　　　　　　在parseArguments方法中，通过读取命令行参数，然后把它写入配置文件，在这边又从配置文件读出来。

• 判定是否为伪分布式存储参数配置的处理，存储信息读取，恢复，调节和初始化。
  
• 利用ServerSocket把服务器绑定到datanode的socket地址，并设置接收缓存大小为128K。（侦听、接受请求、关闭等都在DataXceiverServer这个类处理，DataNode只负责创建ServerSocket）
  
• 建立dataXceiverServer线程组，this.threadGroup = new ThreadGroup("dataXceiverServer");
  
• 后台运行DataXceiverServer（见另一篇文章《hadoop源代码分析（2）-hdfs.server.datanode包-DataXceiverServer类【原创】》）
  

　　　　　　a、并把它加入threadGroup线程组，同时启动ServerSoceket的accpet（）方法，侦听并接受来自客户端或其他服务器的连接请求。
　　　　　　b、后台运行DataXceiver，读取sockets，并执行DataXceiver.run()方法，读取数据。

• 下面就是初始化基本属性，如块报告间隔时间、块延迟报告时间，心跳间隔时间，这些都是从配置文件获取。
  
• 初始化DataBlockScanner（数据块扫描器）按照如下规则：
  

　　　　　　a、读取dfs.datanode.scan.period.hours，如果值小于0，那么说明verification被关闭，
　　　　　　b、如果data不是FSDataset的实例，那么说明不支持FSDataset。
　　　　　　c、如果上面两者都不时，那么就数据块扫描器初始化。

• 创建servlet为http服务，内部用jetty实现，用于对页面的监控。
  

　　　　　　a、用HttpServer构造方法创建http服务器，根据secureResource是否为空采取不同的服务方式；
　　　　　　b、检查https协议是否可用，默认为不可用，若可用，读取客户端权限（默认为不设权限），https地址，新增ssl配置文件：ssl-server.xml。服务器开始监听ssl。不可用，下一步。
　　　　　　c、把多个sevlets加入服务器（StreamFile，FileChecksumServlets.GetServlet，DataBlockScanner.Servlet.）里举个具体实现的例子：如把StreamFile加入jetty：this.infoServer.addInternalServlet(null, "/streamFile/*", StreamFile.class);代码是将HADOOP_HOME\webapp下面的treamFile目录作为了jetty的默认Context。
　　　　　　d、调用setAttribute方法设置属性，datanode，datanode.blockScanner
　　　　　　e、判断wedhdfs文件系统是否可用，选择把相关的包加入到JerseyResource
　　　　　　 f、启动HttpServer。this.infoServer.start();
　　　　　　 g、把当前服务器端口加入注册信息。
　　　　　　 h、创建DataNode运行时记录信息的对象。myMetrics = DataNodeInstrumentation.create(conf,                                          dnRegistration.getStorageID());
　　　　　　  i、获取服务权限并更新。
　　　　　　  j、初始化slave mode 的blockTokenSecretManager。
　　　　　　  k、开启ipc服务。
　　　　这就是整个DataNode的启动顺序，整个过程还是比较复杂的，特别时要理清为什么每个步骤怎么做，还需要对hadoop框架及流程更加清晰才有可能。　　

2、DataNode运行
　　--------------------------------------------------待续-----------------

3、DataNode通信
　　--------------------------------------------------待续-----------------

4、DataNode数据块存储
　　--------------------------------------------------待续-----------------

5、DataNode安全保证
　　--------------------------------------------------待续-----------------

四  DataNode主要方法、属性分析

A：属性
　　　　1、datanode给namenode报告数据块的时间。



1  /**
2     * 当数据块报告要花费很长时间时，开始记录日志时间的初始值，在硬盘高负载和内存压力下，
3 　   *几分钟的数据块报告是正常现象，因为他们会引发很多硬盘检索。
4     */  
5    private static final long LATE_BLOCK_REPORT_WARN_THRESHOLD =
6        10 * 60 * 1000; // 10m
7    private static final long LATE_BLOCK_REPORT_INFO_THRESHOLD =
8        3 * 60 * 1000; // 3m
9 // 正常情况下3min开始记录日志，警告时，10min开始记录日志
　　　　2、其他一些属性　


View Code


  /**
* 数据节点协议
*/
public DatanodeProtocol namenode = null;
/**
* 在数据节点上存储数据块的磁盘接口（分为分布式和伪分布式）
*/
public FSDatasetInterface data = null;
/**
* 数据节点注册信息
*/
public DatanodeRegistration dnRegistration = null;
volatile boolean shouldRun = true;
/**
* 记录DN接受到的数据块的链表，通过该数据结构向NN报告接收到的block的信息
*/
private LinkedList receivedBlockList = new LinkedList();
/**
* 后台运行数据块接收服务器，启动ServerSoceket的监听以及接受
*/  
Daemon dataXceiverServer = null;
/**
* java.lang中的线程组
*/
ThreadGroup threadGroup = null;
/**
* 块报告间隔时间
*/
long blockReportInterval;
//disallow the sending of BR before instructed to do so
long lastBlockReport = 0;
boolean resetBlockReportTime = true;
/**
* 块报告延迟时间
*/
long initialBlockReportDelay = BLOCKREPORT_INITIAL_DELAY * 1000L;
long lastHeartbeat = 0;
/**
* 心跳间隔时间
*/
long heartBeatInterval;
private DataStorage storage = null;
/**
* HttpServer：创建一个能用jetty服务器响应http请求。
*/
private HttpServer infoServer = null;
/**
* 记录DN的运行过程中的一些信息
*/
DataNodeInstrumentation myMetrics;
/**
* 主要是DataNode和DataNode之间recover block时使用
* 管理不同DN之间进行数据传输
*/
public Server ipcServer;
　　　　3、DataNode用到的配置文件参数：

参数	表示含义
dfs.https.enable	是否启用https服务
dfs.https.need.client.auth	https服务是否需要检查客户端权限，默认为false
dfs.datanode.https.address	https地址
dfs.datanode.scan.period.hours	数据块扫描阶段间隔时间
dfs.heartbeat.interval	心跳报告间隔时间（设置时用s表示，内部转换为ms）
dfs.blockreport.initialDelay	初始化数据块报告延迟时间设置时用s表示，内部转换为ms）
dfs.blockreport.intervalMsec	数据块报告间隔时间（ms）
dfs.datanode.simulateddatastorage	是否伪分布式存储（true/flase）
dfs.datanode.transferTo.allowed	是否允许传输（true/false）
dfs.datanode.socket.write.timeout	套接字写入超时时间（ms）
dfs.socket.timeout	套接字超时时间（ms）
dfs.datanode.dns.interface
dfs.datanode.dns.nameserver
slave.host.name	slave主机名
dfs.support.append
dfs.datanode.artificialBlockReceivedDelay	允许推迟发送blockReceived RPCs信号时间
dfs.https.need.client.auth
dfs.https.server.keystore.resource	ssl认证的资源，ssl-server.xml
dfs.datanode.https.address
dfs.https.enable
hadoop.security.authorization	hadoop权限认证
dfs.datanode.ipc.address	ipc地址
dfs.datanode.handler.count	datannode节点的指令信息

　　

B：方法
　　　 　1、注册数据节点的 MXBean，用Mbean.Register()。Mbean描述一个可管理的资源，是一个java对象，必须是共用的，非抽象的类，至少有一个共用的 构造器，必须实现它自己的相应的MBean接口。注册MBean：MBeanServer的主要职责时在一个JMX代理中维护一个MBean的注册。 MBean是JMX（Java Management Extensions）针对每一个需要管理的资源创建的实例时JMX框架所要求的。（不是很理解）


View registerMXBean Code


1   private ObjectName mxBean = null;
2   /**
3    * Register the DataNode MXBean using the name
4    *        "hadoop:service=DataNode,name=DataNodeInfo"
5    */
6   void registerMXBean(Configuration conf) {
7     // We wrap to bypass standard mbean naming convention.
8     // This wraping can be removed in java 6 as it is more flexible in
9     // package naming for mbeans and their impl.
10     mxBean = MBeans.register("DataNode", "DataNodeInfo", this);
11   }
五  DataNode相关类、接口简述
　　　　1、DataBlockScanner：数据块扫描器。
　　　　2、DataXceiverServer：见《hadoop源代码分析（2）-hdfs.server.datanode包-DataXceiverServer类【原创】》。
　　　　3、相关类：DatanodeRegistration，这个类主要用于，当Datanode向Namenode发送注册信息时，它要向Namenode提供一些自己的注册信息。
　　　　4、相关类：ServiceAuthorizationManager
　　　　5、相关类：RPC。这个要系统地学习了hadoop ipc协议后，理解的才比较透彻。

　　　　6、相关类：StreamFile，在方法addInternalServlet(null, "/streamFile/*", StreamFile.class);中，把Stream加入服务器。
　　　　7、相关类：FileChecksumServlets，在方法this.infoServer.addInternalServlet(null, "/getFileChecksum/*",
        FileChecksumServlets.GetServlet.class);中，把FileChecksumServlets加入服务器。
　　　　8、相关类：DataNodeInstrumentation，记录DataNode运行时相关信息。
　　　　9、相关类：BlockTokenSecretManager， 数据块符号管理器有两种实例模式，master mode和slave mode。master模式能生成新的数据块键并且输出数据块键给slave模式，但是slave模式只能导入或者使用从maste收到的数据块键。这两种模式都能都生成和修改数据块符号。master mode--NN，slave mode--DN。
　　　　10、实现InterDatanodeProtocol, ClientDatanodeProtocol, FSConstants, Runnable, DataNodeMXBean等接口。FSConstants定义跟文件系统相关的常量。
　　　　11、相关类HttpServer。HttpServer类用来创建一个内嵌的Jetty服务器来响应http请求。首要目标是为服务器提供状态信息。有三种上下文方式："/logs/" ->指向日志目录；“/static/" -> 指向(src/webapps/static)；" / " -> the jsp server code from (src/webapps/)；

六  结语
　　　　原文出处：http://www.iyunv.com/caoyuanzhanlang
　　　　

草原战狼淘宝小店：http://xarxf.taobao.com/ 淘宝搜小矮人鞋坊，主营精致美丽时尚女鞋，为您的白雪公主挑一双哦。谢谢各位博友的支持。
　　==========================================================================================================
　　　　===================================    以上分析仅代表个人观点，欢迎指正与交流   ===================================
　　　　===================================    尊重劳动成果，转载请注明出处，万分感谢   ===================================
　　　　==========================================================================================================