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中。

原文地址:https://www.cnblogs.com/gdy1993/p/9379872.html

时间: 2024-11-09 00:09:15

hadoop之hdfs------------------FileSystem及其源码分析的相关文章

Hadoop FileInputFormat实现原理及源码分析

FileInputFormat(org.apache.hadoop.mapreduce.lib.input.FileInputFormat)是专门针对文件类型的数据源而设计的,也是一个抽象类,它提供两方面的作用: (1)定义Job输入文件的静态方法: (2)为输入文件形成切片的通用实现: 至于如何将切片中的数据转换为一条条的“记录”则根据文件类型的不同交由具体的子类负责实现. FileInputFormat input paths FileInputFormat提供了四个静态方法用于定义Job的

【JAVA集合】LinkedHashMap及其源码分析

以下内容基于jdk1.7.0_79源码: 什么是LinkedHashMap 继承自HashMap,一个有序的Map接口实现,这里的有序指的是元素可以按插入顺序或访问顺序排列: LinkedHashMap补充说明 与HashMap的异同:同样是基于散列表实现,区别是,LinkedHashMap内部多了一个双向循环链表的维护,该链表是有序的,可以按元素插入顺序或元素最近访问顺序(LRU)排列, 简单地说:LinkedHashMap=散列表+循环双向链表 LinkedHashMap的数组结构 用画图工

[Java]I/O底层原理之一:字符流、字节流及其源码分析

关于 I/O 的类可以分为四种: 关于字节的操作:InputStream 和 OutPutStream: 关于字符的操作:Writer 和 Reader: 关于磁盘的操作:File: 关于网络的操作:Socket( Socket 类不在 java.io 包中). 在本篇博客中我们来看一下前两种 I/O,即字符流与字节流,首先两者的实现关系如下图所示 一.字节流 在字节流的类中,最顶层的是 Inputstream 抽象类和 OutputStream 抽象类,两者定义了一些关于字节数据读写的基本操作

Hadoop 1.x的Shuffle源码分析之3

shuffle有两种,一种是在内存存储数据,另一种是在本地文件存储数据,两者几乎一致. 以本地文件进行shuffle的过程为例: mapOutput = shuffleToDisk(mapOutputLoc, input, filename, compressedLength) shuffleToDisk函数如下: private MapOutput shuffleToDisk(MapOutputLocation mapOutputLoc, InputStream input, Path fil

hadoop 2.6.0 LightWeightGSet源码分析

LightWeightGSet的作用用一个数组来存储元素,而且用链表来解决冲突.不能rehash.所以内部数组永远不用改变大小.此类不支持空元素. 此类也不是线程安全的.有两个类型參数.第一个用于查找元素,第二个类型參数必须是第一个类型參数的子类,而且必须实现LinkedElement接口. /** * A low memory footprint {@link GSet} implementation, * which uses an array for storing the element

Hadoop之HDFS原理及文件上传下载源码分析(上)

HDFS原理 首先说明下,hadoop的各种搭建方式不再介绍,相信各位玩hadoop的同学随便都能搭出来. 楼主的环境: 操作系统:Ubuntu 15.10 hadoop版本:2.7.3 HA:否(随便搭了个伪分布式) 文件上传 下图描述了Client向HDFS上传一个200M大小的日志文件的大致过程: 首先,Client发起文件上传请求,即通过RPC与NameNode建立通讯. NameNode与各DataNode使用心跳机制来获取DataNode信息.NameNode收到Client请求后,

深入源码分析Handler的消息处理机制

学习Android的同学注意了!!! 学习过程中遇到什么问题或者想获取学习资源的话,欢迎加入Android学习交流群,群号码:364595326  我们一起学Android! handler的消息处理有三个核心类:Looper,Handler和Message.其实还有一个Message Queue(消息队列),但是MessageQueue被封装到Looper里面了,我们不会直接与MessageQueue打交道,因此我没将其作为核心类.下面一一介绍: 线程的魔法师Looper Looper的字面意

Android -- 消息处理机制源码分析(Looper,Handler,Message)

android的消息处理有三个核心类:Looper,Handler和Message.其实还有一个Message Queue(消息队列),但是MQ被封装到Looper里面了,我们不会直接与MQ打交道,因此我没将其作为核心类.下面一一介绍: Looper Looper的字面意思是“循环者”,它被设计用来使一个普通线程变成Looper线程.所谓Looper线程就是循环工作的线程.在程序开发中(尤其是GUI开发中),我们经常会需要一个线程不断循环,一旦有新任务则执行,执行完继续等待下一个任务,这就是Lo

【转】android的消息处理机制(图+源码分析)——Looper,Handler,Message

原文地址:http://www.cnblogs.com/codingmyworld/archive/2011/09/12/2174255.html#!comments 作为一个大三的预备程序员,我学习android的一大乐趣是可以通过源码学习google大牛们的设计思想.android源码中包含了大量的设计模式,除此以外,android sdk还精心为我们设计了各种helper类,对于和我一样渴望水平得到进阶的人来说,都太值得一读了.这不,前几天为了了解android的消息处理机制,我看了Loo