hadoop之hdfs------------------FileSystem及其源码分析

Posted sowhat1943

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了hadoop之hdfs------------------FileSystem及其源码分析相关的知识,希望对你有一定的参考价值。

FileSystem及其源码分析

  FileSystem这个抽象类提供了丰富的方法用于对文件系统的操作,包括上传、下载、删除、创建等。这里多说的文件系统通常指的是HDFS(DistributedFileSystem),其实,hadoop处理支持分布式文件系统,还提供了对诸如本地文件系统(LocalFileSystem)、FTP文件系统(FTPFIle)的支持。

  在这里我们主要介绍一下DistributedFileSystem的创建过程。如下代码:

  主要包括两个阶段:

    1. 加载配置文件

    2. 初始化文件系统

Configuration conf = new Configuration();//加载配置文件
FileSystem fs = FileSystem.get(conf);//初始化文件系统

  首先来看一下配置文件加载阶段。

  这是Configuration类的静态代码块,默认加载core-default.xml和core-site.xml这两个配置文件。

static{
    //print deprecation warning if hadoop-site.xml is found in classpath
    ClassLoader cL = Thread.currentThread().getContextClassLoader();
    if (cL == null) {
      cL = Configuration.class.getClassLoader();
    }
    if(cL.getResource("hadoop-site.xml")!=null) {//确保在类路径下不存在hadoop-site.xml(已过时)
      LOG.warn("DEPRECATED: hadoop-site.xml found in the classpath. " +
          "Usage of hadoop-site.xml is deprecated. Instead use core-site.xml, "
          + "mapred-site.xml and hdfs-site.xml to override properties of " +
          "core-default.xml, mapred-default.xml and hdfs-default.xml " +
          "respectively");
    }
    addDefaultResource("core-default.xml");
    addDefaultResource("core-site.xml");
  }

  接下来进入到初始化文件系统阶段:

  FileSystem的get(Configuration conf)方法调用了它的另一个方法get(getDefaultUri(conf),conf),这个方法通过判断是否采用了缓存机制,如果采用了缓存机制,则从缓存中获取,如果没有采用缓存机制,则创建新的文件系统,默认开启缓存机制。

 public static FileSystem get(Configuration conf) throws IOException {
    return get(getDefaultUri(conf), conf);
  }
  public static URI getDefaultUri(Configuration conf) {
    return URI.create(fixName(conf.get(FS_DEFAULT_NAME_KEY, DEFAULT_FS)));//通过conf中的fs.defaultFS属性获得URI(hdfs://s101)
  }
public static FileSystem get(URI uri, Configuration conf) throws IOException {
    String scheme = uri.getScheme();//hdfs
    String authority = uri.getAuthority();//s101

    if (scheme == null && authority == null) {     // use default FS :默认为本地文件系统 file:///
      return get(conf);
    }

  //省略部分代码
  //判断是否缓存FileSystem   String disableCacheName = String.format("fs.%s.impl.disable.cache", scheme);   //如果不采用缓存机制,每次都创建新的FileSystem if (conf.getBoolean(disableCacheName, false)) { return createFileSystem(uri, conf); }   //如果采用缓存机制,则从CACHE中获取 return CACHE.get(uri, conf);

  先让我们来看一下这个CACHE到底是个什么东西?

  CACHE是FileSystem的一个静态内部类,内部维护一个HashMap<Key,FileSystem>(FileSystem容器),键为Key类型,Key是CACHE的一个静态内部类,内部封装了Schema(协议,这里指hdfs)、Authority(权限主机,这里指s101),Vaule就是缓存的文件系统。

 static class Cache {
  //省略......
    private final Map<Key, FileSystem> map = new HashMap<Key, FileSystem>();//FileSystem容器
    
    /** FileSystem.Cache.Key */
    static class Key {
      final String scheme;//hdfs
      final String authority;//s101
      final UserGroupInformation ugi;
      //省略...  
      }    
}

  CACHE.get(uri,conf)方法用于获得具体的FileSystem

FileSystem get(URI uri, Configuration conf) throws IOException{
      Key key = new Key(uri, conf);
      return getInternal(uri, conf, key);
    }

  调用getInterval(uri,conf.key)方法:该方法通过createFileSystem创建新的文件系统,并将其存入缓存容器map中。

private FileSystem getInternal(URI uri, Configuration conf, Key key) throws IOException{
      FileSystem fs;
      synchronized (this) {
        fs = map.get(key);//由于此时为一次创建FileSystem,所以此时map为null
      }
      if (fs != null) {
        return fs;
      }
      //fs不为null,创建文件系统
      fs = createFileSystem(uri, conf);
    
      synchronized (this) { // refetch the lock again
        FileSystem oldfs = map.get(key);
        if (oldfs != null) { // a file system is created while lock is releasing
          fs.close(); // close the new file system
          return oldfs;  // return the old file system
        }
        
        // now insert the new file system into the map
        if (map.isEmpty()
                && !ShutdownHookManager.get().isShutdownInProgress()) {
          ShutdownHookManager.get().addShutdownHook(clientFinalizer, SHUTDOWN_HOOK_PRIORITY);
        }
        fs.key = key;
        //将文件系统存入map容器
        map.put(key, fs);
        if (conf.getBoolean("fs.automatic.close", true)) {
          toAutoClose.add(key);
        }
        return fs;
      }
    }

 下面我们来看一下 createFileSystem(uri, conf)是如何创建FileSystem的:

private static FileSystem createFileSystem(URI uri, Configuration conf
      ) throws IOException {
    //根据conf和Schema获取对应的FileSystemClass,这里指的是DistributedFileSystem.class
    Class<?> clazz = getFileSystemClass(uri.getScheme(), conf);
    //通过反射创建文件系统
    FileSystem fs = (FileSystem)ReflectionUtils.newInstance(clazz, conf);
    //初始化文件系统
    fs.initialize(uri, conf);
    return fs;
  }

 简单了解一下getFileSystemClass的获取过程:加载

public static Class<? extends FileSystem> getFileSystemClass(String scheme,
      Configuration conf) throws IOException {
    if (!FILE_SYSTEMS_LOADED) {
      loadFileSystems();//加载文件系统,加载了hadoop支持的9种文件系统,存放到了SERVICE_FILE_SYSTEMS=new HashMap<String,Class<? extends FileSystem>>
}
    Class<? extends FileSystem> clazz = null;
    //省略
    if (clazz == null) {//根据schema获得对应的文件系统的clazz
      clazz = SERVICE_FILE_SYSTEMS.get(scheme);
    }
    if (clazz == null) {
      throw new IOException("No FileSystem for scheme: " + scheme);
    }
    return clazz;
  }

 再来看一下文件系统的initialize()方法做了些什么,最主要的就是创建了DFS客户端对象,是一个DFSClient,它负责与namenode进行远程通信,是一个绝对重要的家伙。

  public void initialize(URI uri, Configuration conf) throws IOException {
        super.initialize(uri, conf);
        this.setConf(conf);
        String host = uri.getHost();
        if (host == null) {
            throw new IOException("Incomplete HDFS URI, no host: " + uri);
        } else {
            this.homeDirPrefix = conf.get("dfs.user.home.dir.prefix", "/user");
            //创建DFS客户端(每个文件系统都持有一个dfs客户端对象)
            this.dfs = new DFSClient(uri, conf, this.statistics);
            this.uri = URI.create(uri.getScheme() + "://" + uri.getAuthority());
            //工作空间
            this.workingDir = this.getHomeDirectory();
        }

到此为止,FileSystem的创建过程就完成了,下面做一下总结。

FileSystem 的创建过程:

  1. 首先加载配置文件,主要是获得fs.defaultFS的属性值。

  2. 创建文件系统:

    首先从CACHE.map缓存中获得相应的文件系统。

    如果是第一次创建该文件系统,加载相应的文件系统的Class对象,通过反射创建文件系统对象,然后调用initialize()方法对初始化

    并存入CACHE.map中。

 





以上是关于hadoop之hdfs------------------FileSystem及其源码分析的主要内容,如果未能解决你的问题,请参考以下文章

Hadoop之hdfs

Hadoop系列Hadoop三大核心之HDFS基础

大数据技术之Hadoop(HDFS)

Hadoop之HDFS介绍

Hadoop 部署之 Hadoop

Hadoop系列之HDFS架构