1. 云图网首页
  2. 技术专区
  3. 大数据

Hadoop2源码分析－HDFS核心模块分析详解大数据

• 大数据

1.概述

　　这篇博客接着《Hadoop2源码分析－RPC机制初识》来讲述，前面我们对MapReduce、序列化、RPC进行了分析和探索，对Hadoop V2的这些模块都有了大致的了解，通过对这些模块的研究，我们明白了MapReduce的运行流程以及内部的实现机制，Hadoop的序列化以及它的通信机制（RPC）。今天我们来研究另一个核心的模块，那就是Hadoop的分布式文件存储系统——HDFS，下面是今天分享的内容目录：

  • HDFS简述
  • NameNode
  • DataNode

　　接下来，我们开始今天的分享内容。

2.HDFS简述

　　HDFS全称Hadoop Distributed File System，在HDFS中有几个基本的概念，首先是它的数据块（Block），HDFS的设计是用于支持大文件的。运行在HDFS上的程序也是用于处理大数据集的。这些程序仅写一次数据，一次或多次读数据请求，并且这些读操作要求满足流式传输速度。HDFS支持文件的一次写多次读操作。HDFS中典型的块大小是64MB，一个HDFS文件可以被被切分成多个64MB大小的块，如果需要，每一个块可以分布在不同的数据节点上。HDFS 中，如果一个文件小于一个数据块的大小，并不占用整个数据块存储空间。

　　HDFS提供了一个可操作文件系统的抽象类org.apache.hadoop.fs.FileSystem，该类被划分在Hadoop-Common部分，其源码地址为：hadoop-2.6.0-src/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/FileSystem.java，如下是FileSystem的部分源码，如下所示：

@InterfaceAudience.Public 
@InterfaceStability.Stable 
public abstract class FileSystem extends Configured implements Closeable { 
        // 代码内容省略 
        // ...             
}

　　我们可以使用着抽象类，去操作HDFS系统上的内容，实现代码如下所示：

private static void dfs() { 
        FileSystem fs = null; 
        try { 
            fs = FileSystem.get(conf);// get file object 
            FileStatus[] list = fs.listStatus(new Path("/"));// file status list 
            for (FileStatus file : list) { 
                LOGGER.info(file.getPath().getName());// print file names 
            } 
        } catch (IOException e) { 
            e.printStackTrace(); 
            LOGGER.error("Get hdfs path has error,msg is " + e.getMessage()); 
        } finally { 
            try { 
                if (fs != null) { 
                    fs.close(); 
                } 
            } catch (IOException e) { 
                e.printStackTrace(); 
                LOGGER.error("Close fs object has error,msg is " + e.getMessage()); 
            } 
        } 
    }

　　下面，我们来看另一个概念是元数据节点(Namenode)和数据节点(datanode)，这2个是HDFS的核心模块，下面我们分别来看看这2个核心模块。

3.NameNode

　　NN节点用来管理文件系统的NameSpace，将所有的文件和文件夹的Meta保存在一个文件系统中，是HDFS中文件目录和文件分配的管理者，保存的重要信息如下所示：

Hadoop2源码分析－HDFS核心模块分析详解大数据

　　在HDFS集群上可能包含成百上千个DataNode（简称DN）节点，这些DN节点定时和NameNode（简称NN）节点保持通信，接受NN节点的一些指令，为了减小NN的压力，NN上并不永久存储那个DN上报的数据块信息，而是通过DN上报的状态来更新NN上的映射表信息。DN和NN建立连接后，会和NN保持心跳，心跳返回的信息包含了NN对DN的一些指令信息，如删除数据，复制数据到其他的DN节点。值得注意的是NN不会主动去请求DN，这是一个严格意义上的C/S架构模型，同时，客户端在操作HDFS集群时，DN节点会互相配合，保证数据的一致性。

　　NN节点信息存储，部分截图信息如下所示：

Hadoop2源码分析－HDFS核心模块分析详解大数据

Hadoop2源码分析－HDFS核心模块分析详解大数据

4.DataNode

　　下面我们来分析一下DN的实现，DN的实现包含以下部分，一部分是对本地Block的管理，另一部分就是和其他的Entity进行数据交互。首先，我们先看本地的Block管理部分。我们在搭建Hadoop集群时，会指定Block的存储路径，我们可以找到配置的存储路径，在hdfs-site.xml文件下，内容路径如下所示：

<property> 
        <name>dfs.datanode.data.dir</name> 
        <value>/home/hadoop/data/dfs/data</value> 
</property>

　　然后，我们进入到DN节点上，找到对应的存储目录，如下图所示：

Hadoop2源码分析－HDFS核心模块分析详解大数据

　　这里面in_use.lock的作用是做一个排斥操作，在对应的应用上面加锁。然后current目录存放的是当前有效的Block，进入到current目录后，出现如下图所示的目录：Hadoop2源码分析－HDFS核心模块分析详解大数据

　　VERSION存放着一些文件的Meta，接着还有一系列的Block文件和Meta文件，Block文件是存储了HDFS中的数据的。存储的Block，一个Block在多个DN节点上有备份，其备份参数可以调节，在hdfs-site.xml文件中，属性设置如下所示：

<property> 
        <name>dfs.replication</name> 
        <value>3</value> 
</property>

　　首先，我们来看DateNode的类，部分代码如下所示：

@VisibleForTesting 
  @InterfaceAudience.Private 
  public static DataNode createDataNode(String args[], Configuration conf, 
      SecureResources resources) throws IOException { 
    DataNode dn = instantiateDataNode(args, conf, resources);// init dn 
    if (dn != null) { 
      dn.runDatanodeDaemon();// register to nn and back to dn thread 
    } 
    return dn; 
  }
/** Instantiate a single datanode object, along with its secure resources.  
   * This must be run by invoking{@link DataNode#runDatanodeDaemon()}  
   * subsequently.  
   */ 
  public static DataNode instantiateDataNode(String args [], Configuration conf, 
      SecureResources resources) throws IOException { 
    if (conf == null) 
      conf = new HdfsConfiguration(); 
     
    if (args != null) { 
      // parse generic hadoop options 
      GenericOptionsParser hParser = new GenericOptionsParser(conf, args); 
      args = hParser.getRemainingArgs(); 
    } 
     
    if (!parseArguments(args, conf)) { 
      printUsage(System.err); 
      return null; 
    } 
    Collection<StorageLocation> dataLocations = getStorageLocations(conf); 
    UserGroupInformation.setConfiguration(conf); 
    SecurityUtil.login(conf, DFS_DATANODE_KEYTAB_FILE_KEY, 
        DFS_DATANODE_KERBEROS_PRINCIPAL_KEY); 
    return makeInstance(dataLocations, conf, resources); 
  }
static DataNode makeInstance(Collection<StorageLocation> dataDirs, 
      Configuration conf, SecureResources resources) throws IOException { 
    LocalFileSystem localFS = FileSystem.getLocal(conf); 
    FsPermission permission = new FsPermission( 
        conf.get(DFS_DATANODE_DATA_DIR_PERMISSION_KEY, 
                 DFS_DATANODE_DATA_DIR_PERMISSION_DEFAULT)); 
    DataNodeDiskChecker dataNodeDiskChecker = 
        new DataNodeDiskChecker(permission); 
    List<StorageLocation> locations = 
        checkStorageLocations(dataDirs, localFS, dataNodeDiskChecker); 
    DefaultMetricsSystem.initialize("DataNode"); 
 
    assert locations.size() > 0 : "number of data directories should be > 0"; 
    return new DataNode(conf, locations, resources);// create dn obejct 
  }
public void runDatanodeDaemon() throws IOException { 
    blockPoolManager.startAll(); 
 
    // start dataXceiveServer 
    dataXceiverServer.start(); 
    if (localDataXceiverServer != null) { 
      localDataXceiverServer.start(); 
    } 
    ipcServer.start(); 
    startPlugins(conf); 
  }
public static void secureMain(String args[], SecureResources resources) { 
    int errorCode = 0; 
    try { 
      StringUtils.startupShutdownMessage(DataNode.class, args, LOG); 
      DataNode datanode = createDataNode(args, null, resources); 
      if (datanode != null) { 
        datanode.join(); 
      } else { 
        errorCode = 1; 
      } 
    } catch (Throwable e) { 
      LOG.fatal("Exception in secureMain", e); 
      terminate(1, e); 
    } finally { 
      // We need to terminate the process here because either shutdown was called 
      // or some disk related conditions like volumes tolerated or volumes required 
      // condition was not met. Also, In secure mode, control will go to Jsvc 
      // and Datanode process hangs if it does not exit. 
      LOG.warn("Exiting Datanode"); 
      terminate(errorCode); 
    } 
  }
  • Main函数入口

　　下面给出DN类的Main函数入口，代码片段如下所示：

 public static void main(String args[]) { 
    if (DFSUtil.parseHelpArgument(args, DataNode.USAGE, System.out, true)) { 
      System.exit(0); 
    } 
 
    secureMain(args, null); 
  }

5.总结

　　在研究HDFS的相关模块时，这里需要明白各个模块的功能及作用，这里为大家介绍了DN类的部分代码片段，以及给代码片段重要部分添加了代码注释，若是大家需要了解详细的相关流程及代码，可以阅读Hadoop的HDFS部分的源代码。

6.结束语

　　这篇博客就和大家分享到这里，如果大家在研究学习的过程当中有什么问题，可以加群进行讨论或发送邮件给我，我会尽我所能为您解答，与君共勉！

原创文章，作者：奋斗，如若转载，请注明出处：https://blog.ytso.com/9932.html

赞 (0)
0
上一篇 2021年7月19日
下一篇 2021年7月19日

相关推荐

发表回复

登录后才能评论