HDFS下载数据之源码分析-FileSystem.get(conf)_block01

水深无声 2022-01-13 22:29 227阅读 0赞

[2019独角兽企业重金招聘Python工程师标准>>> ][2019_Python_] ![hot3.png][]

首先来看一下, FileSystem(org.apache.hadoop.fs.FileSystem), 这是一个抽象类, 是所有文件系统的父类.

而我们要从HDFS(Hadoop Distributed FileSystem)下载数据, 应该获取一个DistributedFileSystem的实例,那么如何获取一个DistributedFileSystem的实例呢?

FileSystem fs = FileSystem.get(new Configuration());

在FileSystem中有3个重载的get()方法

// 1.通过配置文件获取一个FileSystem实例
    public static FileSystem get(Configuration conf)
    // 2.通过指定的FileSystem的URI, 配置文件获取一个FileSystem实例
    public static FileSystem get(URI uri, Configuration conf)
    // 3.通过指定的FileSystem的URI, 配置文件, FileSystem用户名获取一个FileSystem实例
    public static FileSystem get(final URI uri, final Configuration conf, final String user)

先调用FileSystem.get(Configuration conf)方法,再调用重载方法FileSystem.get(URI uri, Configuration conf)

public static FileSystem get(URI uri, Configuration conf) throws IOException {
        // schem是FileSystem具体的URI方案如: file, hdfs, Webhdfs, har等等
        String scheme = uri.getScheme();    // scheme = hdfs
        // authority是NameNode的主机名, 端口号
        String authority = uri.getAuthority();    // authority = node1:9000
        ...
        // disableCacheName = fs.hdfs.impl.disable.cache
        String disableCacheName = String.format("fs.%s.impl.disable.cache", scheme);    
        // 读取配置文件, 判断是否禁用缓存
        if (conf.getBoolean(disableCacheName, false)) {    // 若禁用缓存
          return createFileSystem(uri, conf);    // 直接调用创建FileSystem实例的方法
        }
        // 不禁用缓存, 先从FileSystem的静态成员变量CACHE中获取FileSystem实例
        return CACHE.get(uri, conf);
    }

再调用FileSystem$Cache.get(URI uri, Configuration conf)方法(Cache是FileSystem的静态内部类)

FileSystem get(URI uri, Configuration conf) throws IOException{
          Key key = new Key(uri, conf);    // key = (root (auth:SIMPLE))@hdfs://node1:9000
          return getInternal(uri, conf, key);
    }

再调用FileSystem$Cache.getInternal(URI uri, Configuration conf, FileSystem$Cache$Key key)方法(Key又是Cache的静态内部类)

private FileSystem getInternal(URI uri, Configuration conf, Key key) throws IOException{
          FileSystem fs;
          synchronized (this) {
            // map是Cache中用来缓存FileSystem实例的成员变量, 其类型为HashMap<Key, FileSystem>
            fs = map.get(key);
          }
          if (fs != null) {    // 如果从缓存map中获取到了相应的FileSystem实例
            return fs;    // 则返回这个实例
          }
          // 否则, 调用FileSystem.createFileSystem(URI uri, Configuration conf)方法, 创建FileSystem实例
          fs = createFileSystem(uri, conf);
          /* 分割线1, 期待着createFileSystem()方法的返回 */
          synchronized (this) { // refetch the lock again
            /*
             * 在多线程环境下, 可能另一个客户端(另一个线程)创建好了一个DistributedFileSystem实例, 并缓存到了map中
             * 所以, 这时候就把当前客户端新创建的DistributedFileSystem实例注销
             * 其实这是一个特殊的单例模式, 一个key映射一个DistributedFileSystem实例
             */
            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
             * 缓存当前新创建的DistributedFileSystem实例到map中
             */
            fs.key = key;
            map.put(key, fs);
            ...
            return fs;
          }
    }

**来自分割线1**, 先调用FileSystem.createFileSystem(URI uri, Configuration conf)方法

private static FileSystem createFileSystem(URI uri, Configuration conf
          ) throws IOException {
        // 通过读取配置文件, 获取FileSystem具体的URI模式: hdfs的类对象
        Class<?> clazz = getFileSystemClass(uri.getScheme(), conf); // clazz = org.apache.hadoop.hdfs.DistributedFileSystem
        ...
        // 反射出一个DistributedFileSystem实例
        FileSystem fs = (FileSystem)ReflectionUtils.newInstance(clazz, conf);
        // 对DistributedFileSystem实例初始化
        fs.initialize(uri, conf);
        return fs;
    }

在调用DistributedFileSystem.initialize(URI uri, Configuration conf)方法之前, 先来看一下DistributedFileSystem类吧.

DistributedFileSystem是抽象类FileSystem的子类实现,

public class DistributedFileSystem extends FileSystem {
      ...
      DFSClient dfs;    // DistributedFileSystem持有一个DFSClient类型的成员变量dfs, 最重要的成员变量!
      ...
    }

调用DistributedFileSystem.initialize(URI uri, Configuration conf)方法

public void initialize(URI uri, Configuration conf) throws IOException {
        ...
        // new一个DFSClient实例, 成员变量dfs引用这个DFSClient实例
        this.dfs = new DFSClient(uri, conf, statistics );
        /* 分割线2, 期待着new DFSClient()的返回 */
        ...
    }

在new DFSClient实例之前, 先来看一下DFSClient类吧! 看一下到底要为哪些成员变量赋值

public class DFSClient implements java.io.Closeable, RemotePeerFactory {
      ...
      final ClientProtocol namenode;    //DFSClient持有一个ClientProtocol类型的成员变量namenode, 一个RPC代理对象
      /* The service used for delegation tokens */
      private Text dtService;
      ...
    }

**来自分割线2**, 调用DFSClient的构造函数DFSClient(URI nameNodeUri, Configuration conf, FileSystem$Statistics statistics), 再调用重载构造函数DFSClient(URI nameNodeUri, ClientProtocol rpcNamenode, Configuration conf, FileSystem$Statistics statistics)

public DFSClient(URI nameNodeUri, ClientProtocol rpcNamenode, Configuration conf, 
        FileSystem.Statistics stats) throws IOException {
        ...
        NameNodeProxies.ProxyAndInfo<ClientProtocol> proxyInfo = null;
        if (numResponseToDrop > 0) {    // numResponseToDrop = 0
          // This case is used for testing.
          LOG.warn(DFSConfigKeys.DFS_CLIENT_TEST_DROP_NAMENODE_RESPONSE_NUM_KEY
              + " is set to " + numResponseToDrop
              + ", this hacked client will proactively drop responses");
          proxyInfo = NameNodeProxies.createProxyWithLossyRetryHandler(conf,
              nameNodeUri, ClientProtocol.class, numResponseToDrop);
        }
        
        if (proxyInfo != null) { // proxyInfo = null
          this.dtService = proxyInfo.getDelegationTokenService();
          this.namenode = proxyInfo.getProxy();
        } else if (rpcNamenode != null) { // rpcNamenode = null
          // This case is used for testing.
          Preconditions.checkArgument(nameNodeUri == null);
          this.namenode = rpcNamenode;
          dtService = null;
        } else {    // 前面两个if只在测试的情况下成立, 这个else的代码块才是重点
          ...
          /*
           * 创建一个NameNodeProxies.ProxyAndInfo<ClientProtocol>类型的对象, proxyInfo引用这个对象 
           * createProxy(conf, nameNodeUri, ClientProtocol.class)方法是不是和RPC.getProxy(Class<T> protocol,
           * long clientVersion, InetSocketAddress addr, Configuration conf)很像?
           * 没错! 你没看错! 这说明createProxy()方法内部一定会调用RPC的相关方法
           * conf    都是Configuration类型的conf
           * nameNodeUri = hdfs://node1:9000    这不就是InetSocketAddress类型的addr的hostName和port
           * ClientProtocol.class    都是RPC protocol接口的类对象
           * ClientProtocol is used by user code via DistributedFileSystem class to communicate 
           * with the NameNode
           * ClientProtocol是DistributedFileSystem用来与NameNode通信的
           */
          proxyInfo = NameNodeProxies.createProxy(conf, nameNodeUri, ClientProtocol.class);
          /* 分割线3, 期待着createProxy()方法的返回 */
          this.dtService = proxyInfo.getDelegationTokenService();
          this.namenode = proxyInfo.getProxy();
        }
        ...
    }

**来自分割线3**, 调用NameNodeProxies.createProxy(Configuration conf, URI nameNodeUri, Class<T> xface)方法

/**
       * Creates the namenode proxy with the passed protocol. This will handle
       * creation of either HA- or non-HA-enabled proxy objects, depending upon
       * if the provided URI is a configured logical URI.
       * 通过传过来的protocol参数, 创建namenode的代理对象. 至于是HA还是非HA的namenode代理对象, 
       * 这取决于实际搭建的Hadoop环境
       **/
    public static <T> ProxyAndInfo<T> createProxy(Configuration conf, URI nameNodeUri, Class<T> xface)
        throws IOException {
        // 获取Hadoop实际环境中HA的配置
        Class<FailoverProxyProvider<T>> failoverProxyProviderClass =
            getFailoverProxyProviderClass(conf, nameNodeUri, xface);
    
        if (failoverProxyProviderClass == null) {    // 非HA,这里是Hadoop的伪分布式搭建
          // Non-HA case, 创建一个非HA的namenode代理对象
          return createNonHAProxy(conf, NameNode.getAddress(nameNodeUri), xface,
              UserGroupInformation.getCurrentUser(), true);
        } else {    // HA
          // HA case
          FailoverProxyProvider<T> failoverProxyProvider = NameNodeProxies
              .createFailoverProxyProvider(conf, failoverProxyProviderClass, xface,
                  nameNodeUri);
          Conf config = new Conf(conf);
          T proxy = (T) RetryProxy.create(xface, failoverProxyProvider,
              RetryPolicies.failoverOnNetworkException(
                  RetryPolicies.TRY_ONCE_THEN_FAIL, config.maxFailoverAttempts,
                  config.maxRetryAttempts, config.failoverSleepBaseMillis,
                  config.failoverSleepMaxMillis));
          
          Text dtService = HAUtil.buildTokenServiceForLogicalUri(nameNodeUri);
          // 返回一个proxy, dtService的封装对象proxyInfo
          return new ProxyAndInfo<T>(proxy, dtService);
        }
    }

调用NameNodeProxies.createNonHAProxy(Configuration conf, InetSocketAddress nnAddr, Class<T> xface, UserGroupInformation ugi, boolean withRetries)方法

public static <T> ProxyAndInfo<T> createNonHAProxy(Configuration conf, InetSocketAddress nnAddr,
        Class<T> xface, UserGroupInformation ugi, boolean withRetries) throws IOException {
        Text dtService = SecurityUtil.buildTokenService(nnAddr);    //dtService = 192.168.8.101:9000
        T proxy;
        if (xface == ClientProtocol.class) {    // xface = ClientProtocol.class
          // 创建一个namenode代理对象
          proxy = (T) createNNProxyWithClientProtocol(nnAddr, conf, ugi, withRetries);
          /* 分割线4, 期待着createNNProxyWithClientProtocol()方法返回 */
        } else if {
          ...
        }
        // 把proxy, dtService封装成一个ProxyAndInfo对象, 并返回
        return new ProxyAndInfo<T>(proxy, dtService);
      }

[**block02戳我**][block02]

转载于:https://my.oschina.net/u/2503731/blog/663705

[2019_Python_]: https://my.oschina.net/u/2663968/blog/3061697
[hot3.png]: /images/20220113/bb106d14be9f49d4be06f257158fc142.png
[block02]: http://my.oschina.net/u/2503731/blog/663702