【软件工程实践】Hive研究-Blog8

2021SC@SDUSC

研究内容介绍

本人负责的是负责的是将查询块QB转换成逻辑查询计划（OP Tree）
如下的代码出自apaceh-hive-3.1.2-src/ql/src/java/org/apache/hadoop/hive/ql/plan中，也就是我的分析目标代码。本周的研究计划是解析mapper文件夹下面的最后两个StatsSource.java文件源码以及StatsSource.java文件的源码。

StatsSource.java文件代码解析

我们附上整个java文件代码


package org.apache.hadoop.hive.ql.plan.mapper;

import java.util.Map;
import java.util.Optional;

import org.apache.hadoop.hive.ql.optimizer.signature.OpTreeSignature;
import org.apache.hadoop.hive.ql.stats.OperatorStats;

public interface StatsSource {

  boolean canProvideStatsFor(Class clazz);

  Optional lookup(OpTreeSignature treeSig);

  void putAll(Map map);

}

非常明显，这是一个接口类的java文件。而对于在本文件下的三个方法：canProvideStatsFor方法、lookup方法、putAll方法，均在先前就已经研究过了，具体内容在Blog3以及Blog4中已经讲解，故我们可以直接引用。
以下的两个连接可以直接查看具体的方法解释。
Blog3的地址以及Blog4的地址

StatsSources.java文件代码解析

我们首先附上整个java文件的源码。


package org.apache.hadoop.hive.ql.plan.mapper;

import java.util.Iterator;
import java.util.List;
import java.util.Map;

import org.apache.hadoop.hive.conf.HiveConf;
import org.apache.hadoop.hive.conf.HiveConf.ConfVars;
import org.apache.hadoop.hive.ql.optimizer.signature.OpTreeSignature;
import org.apache.hadoop.hive.ql.plan.mapper.PlanMapper.EquivGroup;
import org.apache.hadoop.hive.ql.stats.OperatorStats;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.google.common.annotations.VisibleForTesting;
import com.google.common.collect.ImmutableMap;
import com.google.common.collect.ImmutableMap.Builder;

public class StatsSources {

  private static final Logger LOG = LoggerFactory.getLogger(StatsSources.class);

  static enum StatsSourceMode {
    query, hiveserver, metastore;
  }

  public static void initialize(HiveConf hiveConf) {
    // requesting for the stats source will implicitly initialize it
    getStatsSource(hiveConf);
  }

  public static StatsSource getStatsSource(HiveConf conf) {
    String mode = conf.getVar(ConfVars.HIVE_QUERY_REEXECUTION_STATS_PERSISTENCE);
    int cacheSize = conf.getIntVar(ConfVars.HIVE_QUERY_REEXECUTION_STATS_CACHE_SIZE);
    int batchSize = conf.getIntVar(ConfVars.HIVE_QUERY_REEXECUTION_STATS_CACHE_BATCH_SIZE);
    switch (mode) {
    case "query":
      return new MapBackedStatsSource();
    case "hiveserver":
      return StatsSources.globalStatsSource(cacheSize);
    case "metastore":
      return StatsSources.metastoreBackedStatsSource(cacheSize, batchSize, StatsSources.globalStatsSource(cacheSize));
    default:
      throw new RuntimeException("Unknown StatsSource setting: " + mode);
    }
  }

  public static StatsSource getStatsSourceContaining(StatsSource currentStatsSource, PlanMapper pm) {
    StatsSource statsSource = currentStatsSource;
    if (currentStatsSource  == EmptyStatsSource.INSTANCE) {
      statsSource = new MapBackedStatsSource();
    }

    Map statMap = extractStatMapFromPlanMapper(pm);
    statsSource.putAll(statMap);
    return statsSource;
  }

  private static Map extractStatMapFromPlanMapper(PlanMapper pm) {
    Builder map = ImmutableMap.builder();
    Iterator it = pm.iterateGroups();
    while (it.hasNext()) {
      EquivGroup e = it.next();
      List stat = e.getAll(OperatorStats.class);
      List sig = e.getAll(OpTreeSignature.class);

      if (stat.size() > 1 || sig.size() > 1) {
        StringBuffer sb = new StringBuffer();
        sb.append(String.format("expected(stat-sig) 1-1, got {}-{} ;", stat.size(), sig.size()));
        for (OperatorStats s : stat) {
          sb.append(s);
          sb.append(";");
        }
        for (OpTreeSignature s : sig) {
          sb.append(s);
          sb.append(";");
        }
        LOG.debug(sb.toString());
      }
      if (stat.size() >= 1 && sig.size() >= 1) {
        map.put(sig.get(0), stat.get(0));
      }
    }
    return map.build();
  }

  private static StatsSource globalStatsSource;
  private static metastoreStatsConnector metastoreStatsConnector;

  public static StatsSource globalStatsSource(int cacheSize) {
    if (globalStatsSource == null) {
      globalStatsSource = new CachingStatsSource(cacheSize);
    }
    return globalStatsSource;
  }

  public static StatsSource metastoreBackedStatsSource(int cacheSize, int batchSize, StatsSource parent) {
    if (metastoreStatsConnector == null) {
      metastoreStatsConnector = new metastoreStatsConnector(cacheSize, batchSize, parent);
    }
    return metastoreStatsConnector;
  }

  @VisibleForTesting
  public static void clearGlobalStats() {
    if (metastoreStatsConnector != null) {
      metastoreStatsConnector.destroy();
    }
    globalStatsSource = null;
    metastoreStatsConnector = null;
  }

}

枚举型常量集合enum

  static enum StatsSourceMode {
    query, hiveserver, metastore;
  }

java中的这个枚举型集合，用来定义指定的常量，无需再创建新的常量，这样可以保证我们的引用对象是准确无误的。

方法initialize

  public static void initialize(HiveConf hiveConf) {
    // requesting for the stats source will implicitly initialize it
    getStatsSource(hiveConf);
  }

整个方法中只调用了一个getStatsSource方法，那么这是一个什么方法呢？我们从可以直接调用，而不需要任何的接口的形式上猜测，这个getStatsSource方法很大几率是一个内部方法，也就是同样在这个类中。我们浏览上下文发现了这个方法。

方法getStatsSource

  public static StatsSource getStatsSource(HiveConf conf) {
    String mode = conf.getVar(ConfVars.HIVE_QUERY_REEXECUTION_STATS_PERSISTENCE);
    int cacheSize = conf.getIntVar(ConfVars.HIVE_QUERY_REEXECUTION_STATS_CACHE_SIZE);
    int batchSize = conf.getIntVar(ConfVars.HIVE_QUERY_REEXECUTION_STATS_CACHE_BATCH_SIZE);
    switch (mode) {
    case "query":
      return new MapBackedStatsSource();
    case "hiveserver":
      return StatsSources.globalStatsSource(cacheSize);
    case "metastore":
      return StatsSources.metastoreBackedStatsSource(cacheSize, batchSize, StatsSources.globalStatsSource(cacheSize));
    default:
      throw new RuntimeException("Unknown StatsSource setting: " + mode);
    }
  }

首先这里出现了一个完全没有见过的全新类型的变量：hiveconf类。这个hiveconf是一个什么样的类，有什么样的参数和特性？我们通过查阅资料发现hiveconf用于定义HIVE执行上下文的属性(配置参数)，可覆盖覆盖hive-site.xml（hive-default.xml）中的参数值，如用户执行目录、日志打印级别、执行队列等，常用的配置属性如下：

参数名称	参数解释
hive.metastore.warehouse.dir	启动时指定用户目录，不同的用户不同的目录
hive.cli.print.current.db	显示当前数据库
hive.root.logger	输出日志信息
hive.cli.print.header	显示列名称
mapred.job.queue.name	执行队列名称

我们对这些参数，可以使用set指令进行修改：

# 首先启动HIVE
hive
# 然后设置参数
set mapred.job.queue.name=root.default

同样的，上面的命令可以用"hive --hiveconf"命令来实现的：

hive --hiveconf "mapred.job.queue.name=root.default"

当然，这是网络上的一些编程人员总结的内容，我们可以浏览一下官方文档对hiveconf这个类的详细解释：

我们大致浏览一下后续的调用方法，发现所调用的方法几乎全部是在下文出现的方法。我们不妨先把下文的方法全部都解释清楚后，再回过头来解释这个getStatsSource方法。

方法getStatsSourceContaining

  public static StatsSource getStatsSourceContaining(StatsSource currentStatsSource, PlanMapper pm) {
    StatsSource statsSource = currentStatsSource;
    if (currentStatsSource  == EmptyStatsSource.INSTANCE) {
      statsSource = new MapBackedStatsSource();
    }

    Map statMap = extractStatMapFromPlanMapper(pm);
    statsSource.putAll(statMap);
    return statsSource;
  }

我们首先来看一下这个if判断语句的条件。那么上面是EmptyStatsSource.INSTANCE方法呢？这个EmptyStatsSource类又是一个什么样的类？我们已经在Blog2中解析过整个类了，这里我们不妨再放出它的源码。


package org.apache.hadoop.hive.ql.plan.mapper;

import java.util.Map;
import java.util.Optional;

import org.apache.hadoop.hive.ql.optimizer.signature.OpTreeSignature;
import org.apache.hadoop.hive.ql.stats.OperatorStats;

public final class EmptyStatsSource implements StatsSource {

  public static StatsSource INSTANCE = new EmptyStatsSource();

  private EmptyStatsSource() {
  }

  @Override
  public boolean canProvideStatsFor(Class class1) {
    return false;
  }

  @Override
  public Optional lookup(OpTreeSignature treeSig) {
    return Optional.empty();
  }

  @Override
  public void putAll(Map map) {
    throw new RuntimeException("This is an empty source!");
  }

}

可以看得出来，这又是一个接口类，而INSTANCE是它的一个内部自定义变量。那么if语句的判断条件就为：判断传入的变量currentStatsSource是否为EmptyStatsSource的INSTANCE变量。我们来看看判断为true时是如何处理的。我们来看看MapBackedStatsSource()是一个什么方法。根据浏览先前的研究内容，我们发现这是一个我们已经研究过了的java文件，整个文件已经被我们解析完毕。那么，这个MapBackedStatsSource()就是一个构造方法，为的是构造这个类的变量，然后赋予statsSource。接下来走完if的语句后，出现了一个新的方法：extractStatMapFromPlanMapper（）.根据我们的观察，它就在下文的方法中。我们先来看看此方法。

方法extractStatMapFromPlanMapper

  private static Map extractStatMapFromPlanMapper(PlanMapper pm) {
    Builder map = ImmutableMap.builder();
    Iterator it = pm.iterateGroups();
    while (it.hasNext()) {
      EquivGroup e = it.next();
      List stat = e.getAll(OperatorStats.class);
      List sig = e.getAll(OpTreeSignature.class);

      if (stat.size() > 1 || sig.size() > 1) {
        StringBuffer sb = new StringBuffer();
        sb.append(String.format("expected(stat-sig) 1-1, got {}-{} ;", stat.size(), sig.size()));
        for (OperatorStats s : stat) {
          sb.append(s);
          sb.append(";");
        }
        for (OpTreeSignature s : sig) {
          sb.append(s);
          sb.append(";");
        }
        LOG.debug(sb.toString());
      }
      if (stat.size() >= 1 && sig.size() >= 1) {
        map.put(sig.get(0), stat.get(0));
      }
    }
    return map.build();
  }

我们先来看一下首次接触的这个Builder是一个什么样的东西。我们在网上查阅资料，得知了如下信息：我们在构建对象的时候，如果对象属性比较多，我们可以（1）使用一个构造器;（2）一个空的构造器，然后使用setter方法进行设置。使用这些方法时会有冗长的构造函数或者setter方法，有不同参数默认值的构造函数需要多次定义，因此我们可以使用builder来简化代码的简介性。那么，ImmutableMap.builder()又是一个什么样的方法呢？我们查阅资料得知ImmutableMap是一个不可变集合，为什么要创建这种类型的集合是为了保证线程安全以及更有效的利用内存。我们就把它当作一个特殊的集合来看就可以了。

接着出现的Iterator是我们在签名讲过的迭代器，是一个用于访问List和Set类型变量的方法。我们在之前的Blog7中已经解释了这个工具。接下来是一个while循环，而循环终止的条件是it.hasNext（）判断为false，整个来看的意思就是遍历整个传入的参数pm。

我们再来看看，参数pm里面装的是一堆EquivGroup类型的变量。那么这个EquivGroup是什么类呢？我们上Hive的官方API文档中进行查阅，找到了如下资料（资料网址）：

按照官方文档的解释，这个就是一个 Group 可能包含不同种类的事物，它们的目的是相关的一堆变量的集合。好了，我们接着继续。在后面调用了一个getAll的方法，很显然，这是EquivGroup内部封装好的方法，在官方文档也没有给出方法的详情解释，知识告诉我们它的返回类型是List，然后要求传入的参数为Class的类型。我们可以合理的推测这个方法返回的是EquivGroup的一些重要的参数组成的链表，相当于一个getter方法的集合。而跟着的参数OperatorStats以及OpTreeSignature分别代表着运行状态，运行标志的关键信息。

我们接着来到if语句。if语句的判断条件为链表stat以及链表sig只要有一个不为空，就判断为true。然后创建一个字符串缓冲区sb，在这个缓冲区内先设定好指定的格式，每一个数据开头前都会有String.format("expected(stat-sig) 1-1, got {}-{} ;", stat.size(), sig.size())这么一个句子，然后后面跟着stat和sig的链表长度如此的格式。接着的两个for循环就是遍历两个链表了。最后，这个debug方法是为了把sb转换为string字符串后在控制台进行输出。

然后进想到最后一个if语句，这里的判断条件是让stat和sig二者的规模同时至少为1，才能判断为true。进入到处理语句后，将sig和stat的第一个参数作为key和value值传入到map中。

方法最后，返回一个关于map的构造器集合。

方法globalStatsSource

  public static StatsSource globalStatsSource(int cacheSize) {
    if (globalStatsSource == null) {
      globalStatsSource = new CachingStatsSource(cacheSize);
    }
    return globalStatsSource;
  }

在开始本方法前，我们先来看一下内部自定义的两个变量：

  private static StatsSource globalStatsSource;
  private static metastoreStatsConnector metastoreStatsConnector;

首先是这个StatsSource类，我们在本章开头就已经介绍完毕了。接着是这个metastoreStatsConnector类，我们在之前的Blog3中已经研究过了。（链接：Blog3）

回到源码中来，我们先看一下if语句。if语句的判断条件为globalStatsSoure是否为空，如果是则将其值设置为CachingStatsSource（cachesize）。这是一个什么方法？我们不妨浏览一下先前的解析，发现这正是Blog1中解释过的一个类。那么这显然就为实例化这个类的方法，也就是构造方法。我们不妨把整个类的源码放在这里一起观看。


package org.apache.hadoop.hive.ql.plan.mapper;

import java.util.Map;
import java.util.Map.Entry;
import java.util.Optional;

import org.apache.hadoop.hive.ql.exec.Operator;
import org.apache.hadoop.hive.ql.optimizer.signature.OpTreeSignature;
import org.apache.hadoop.hive.ql.stats.OperatorStats;

import com.google.common.cache.Cache;
import com.google.common.cache.CacheBuilder;

public class CachingStatsSource implements StatsSource {


  private final Cache cache;

  public CachingStatsSource(int cacheSize) {
    cache = CacheBuilder.newBuilder().maximumSize(cacheSize).build();
  }

  public void put(OpTreeSignature sig, OperatorStats opStat) {
    cache.put(sig, opStat);
  }

  @Override
  public Optional lookup(OpTreeSignature treeSig) {
    return Optional.ofNullable(cache.getIfPresent(treeSig));
  }

  @Override
  public boolean canProvideStatsFor(Class clazz) {
    if (cache.size() > 0 && Operator.class.isAssignableFrom(clazz)) {
      return true;
    }
    return false;
  }

  @Override
  public void putAll(Map map) {
    for (Entry entry : map.entrySet()) {
      put(entry.getKey(), entry.getValue());
    }
  }

}

方法CachingStatsSource

  public CachingStatsSource(int cacheSize) {
    cache = CacheBuilder.newBuilder().maximumSize(cacheSize).build();
  }

这里面包含了一句语法：CacheBuilder.newBuilder().maximunSize(cacheSize).build();
它的作用是什么？经过查阅，我得知cache是用来比如对于同样一次输入不止一次的取值操作时，所使用的缓存。那么这个参数cacheSize就非常显而易见了，它表示的是缓存的容量，当超出maximumSize时就会被回收。相关的，这个maximumSize（）方法就是设置最大缓存容量的方法了，也就是说，这整个方法的作用就是设置一个新的缓存块，由传入的参数cacheSize决定该缓存块的大小，之后的所有操作都是基于该缓存块基础上的。

回到源码中来，这里构造出一个新的CachingStatsSource对象并将其赋予globalStatsSource。方法最后，将globalStatsSource返回。

方法metatoreBackedStatsSource

  public static StatsSource metastoreBackedStatsSource(int cacheSize, int batchSize, StatsSource parent) {
    if (metastoreStatsConnector == null) {
      metastoreStatsConnector = new metastoreStatsConnector(cacheSize, batchSize, parent);
    }
    return metastoreStatsConnector;
  }

这个方法和上个方法是一模一样的，而这个metastoreStatsConnector是我们在Blog3中研究过的文件，因此对于if语句里面的 metastoreStatsConnector = new metastoreStatsConnector(cacheSize, batchSize, parent);，它是一个构造方法。由于这个文件过长，我们不在这里展示全部内容，如想观看可以转至Blog3中进行观看，在这里我们只展现构造方法。

类构造器方法metastoreStatsConnector

  metastoreStatsConnector(int cacheSize, int batchSize, StatsSource ss) {
    this.ss = ss;
    executor = Executors.newSingleThreadExecutor(
        new BasicThreadFactory.Builder()
            .namingPattern("metastore-RuntimeStats-Loader-%d")
            .daemon(true)
            .build());

    executor.submit(new RuntimeStatsLoader(cacheSize, batchSize));
  }

对于这个方法的详细解析，我们可以可以观看先前的解析。

我们首先从头开始观察整个类的作用是什么，我们注意到类的声明开头有如下语句：class metastoreStatsConnector implements StatsSource ，我们注意到关键词implements，这表示整个metastoreStatsConnector.java文件是接口StatsSource的实现类。我们又注意到一行注释：Decorates a StatSource to be loaded and persisted in the metastore as well.，这是开发人员所标注的内容，它的意思为封装要加载的StatSource对象，使其更加持久化到metastore中。这说明了整个类的目标就为该注释所要完成的事情。

我们回到这个类构造方法中来。我们对暂未使用到的私有变量ss不感兴趣，我们关注后续的语句。我们首先看一下方法Executors.newSingleThreadExecutor（）是一个什么方法。经过查阅资料得知，该方法的目的是创建一个单线程化的线程池，它只会用唯一的工作线程来执行任务，保证所有任务按照指定顺序(FIFO, LIFO, 优先级)执行。我们可以阅读该方法的源码：

public static ExecutorService newSingleThreadExecutor() {
    return new FinalizableDelegatedExecutorService
    (new gc(1, 1,
            0L, TimeUnit.MILLISECONDS,
            new linkedBlockingQueue < Runnable > ()));
}

我们下面使用一个案例来直观的了解这个方法：

package com.zhangxueliang.demo.springbootdemo.JUC.c_026_01_ThreadPool;
 
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
 

public class T07_SingleThreadPool {
 
    public static void main(String[] args) {
        ExecutorService executorService = Executors.newSingleThreadExecutor();
        for (int i = 0; i < 6; i++) {
            final int j=i;
            executorService.execute(()->{
                System.out.println(Thread.currentThread().getName()+" "+j);
            });
        }
        executorService.shutdown();
    }
}

如下为输出结果：

pool-1-thread-1 0
pool-1-thread-1 1
pool-1-thread-1 2
pool-1-thread-1 3
pool-1-thread-1 4
pool-1-thread-1 5

我们再来关注方法里面的BasicThreadFactory.builder()这些方法。方法namingPattern里面跟着的参数就是想要设置的线程名称。而对于后面的.daemo()方法，我们得先了解什么是守护线程：
定义：守护线程–也称“服务线程”，在没有用户线程可服务时会自动离开。
优先级：守护线程的优先级比较低，用于为系统中的其它对象和线程提供服务。

我们举一个经典的例子：垃圾回收线程就是一个经典的守护线程，当我们的程序中不再有任何运行的线程时，程序就不会再产生垃圾，垃圾回收器也就无事可做，所以当垃圾回收线程是JVM上仅剩的线程时，垃圾回收线程就会自动离开，而它始终在低级别的状态中运行，用于实时监控和管理系统中可回收资源。

回到源码，这个方法的意思就是传入true或者false，是否将这个要产生的线程设置为守护线程。在源码中传入的参数为true，那么该方法产生的线程就为守护线程。

我们再来关注构造方法最后的executor.submit()语句。显然，我们要关注的是submit()方法，我们查阅资料得知该方法是一个执行任务的方法，并且会有一个执行结果的返回。我们可以使用一个例子来描述该方法：

import java.util.ArrayList;  
import java.util.List;  
import java.util.Random;  
import java.util.concurrent.Callable;  
import java.util.concurrent.ExecutionException;  
import java.util.concurrent.ExecutorService;  
import java.util.concurrent.Executors;  
import java.util.concurrent.Future;  
  
public class ExecutorServiceTest {  
    public static void main(String[] args) {  
        ExecutorService executorService = Executors.newCachedThreadPool();  
        List> resultList = new ArrayList>();  
  
        // 创建10个任务并执行  
        for (int i = 0; i < 10; i++) {  
            // 使用ExecutorService执行Callable类型的任务，并将结果保存在future变量中  
            Future future = executorService.submit(new TaskWithResult(i));  
            // 将任务执行结果存储到List中  
            resultList.add(future);  
        }  
        executorService.shutdown();  
  
        // 遍历任务的结果  
        for (Future fs : resultList) {  
            try {  
                System.out.println(fs.get()); // 打印各个线程（任务）执行的结果  
            } catch (InterruptedException e) {  
                e.printStackTrace();  
            } catch (ExecutionException e) {  
                executorService.shutdownNow();  
                e.printStackTrace();  
                return;  
            }  
        }  
    }  
}  
  
class TaskWithResult implements Callable {  
    private int id;  
  
    public TaskWithResult(int id) {  
        this.id = id;  
    }  
  
      
    public String call() throws Exception {  
        System.out.println("call()方法被自动调用,干活！！！             " + Thread.currentThread().getName());  
        if (new Random().nextBoolean())  
            throw new TaskException("Meet error in task." + Thread.currentThread().getName());  
        // 一个模拟耗时的操作  
        for (int i = 999999999; i > 0; i--)  
            ;  
        return "call()方法被自动调用，任务的结果是：" + id + "    " + Thread.currentThread().getName();  
    }  
}  
  
class TaskException extends Exception {  
    public TaskException(String message) {  
        super(message);  
    }  
}

该例子的输出结果为：

call()方法被自动调用,干活！！！             pool-1-thread-1  
call()方法被自动调用,干活！！！             pool-1-thread-2  
call()方法被自动调用,干活！！！             pool-1-thread-3  
call()方法被自动调用,干活！！！             pool-1-thread-5  
call()方法被自动调用,干活！！！             pool-1-thread-7  
call()方法被自动调用,干活！！！             pool-1-thread-4  
call()方法被自动调用,干活！！！             pool-1-thread-6  
call()方法被自动调用,干活！！！             pool-1-thread-7  
call()方法被自动调用,干活！！！             pool-1-thread-5  
call()方法被自动调用,干活！！！             pool-1-thread-8  
call()方法被自动调用，任务的结果是：0    pool-1-thread-1  
call()方法被自动调用，任务的结果是：1    pool-1-thread-2  
java.util.concurrent.ExecutionException: com.cicc.pts.TaskException: Meet error in task.pool-1-thread-3  
    at java.util.concurrent.FutureTask$Sync.innerGet(FutureTask.java:222)  
    at java.util.concurrent.FutureTask.get(FutureTask.java:83)  
    at com.cicc.pts.ExecutorServiceTest.main(ExecutorServiceTest.java:29)  
Caused by: com.cicc.pts.TaskException: Meet error in task.pool-1-thread-3  
    at com.cicc.pts.TaskWithResult.call(ExecutorServiceTest.java:57)  
    at com.cicc.pts.TaskWithResult.call(ExecutorServiceTest.java:1)  
    at java.util.concurrent.FutureTask$Sync.innerRun(FutureTask.java:303)  
    at java.util.concurrent.FutureTask.run(FutureTask.java:138)  
    at java.util.concurrent.ThreadPoolExecutor$Worker.runTask(ThreadPoolExecutor.java:886)  
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:908)  
    at java.lang.Thread.run(Thread.java:619)

而对于sumbit方法里面的new RuntimeStatsLoader(cacheSize, batchSize)语句，我从需要的参数cacheSize(缓存规模)和batchSize(批处理规模)中可以猜测这个RuntimeStatsLoader类是一个任务处理的类，需要指定缓存和批处理规模来执行任务。在下文我们会见到关于这个类的详细介绍。

至此，整个方法的目的也清楚明了了：创建一个安全的线程并且自定义其名称，方便调用且设置其为守护线程，完成资源自动回收的功能，并且给定缓存规模和批处理规模并执行任务，得到任务执行的结果（包括是否执行成功等信息）。

现在我们回到源码中来，我们清楚这个方法是为了初始化为空的变量，以防止错误和空引用。在方法的最后和上个方法一样，返回初始化的类。

方法clearGlobalStats

  @VisibleForTesting
  public static void clearGlobalStats() {
    if (metastoreStatsConnector != null) {
      metastoreStatsConnector.destroy();
    }
    globalStatsSource = null;
    metastoreStatsConnector = null;
  }

从方法的名字上看就可以看得出来这是一个关闭接口释放资源的方法。我们看看if语句后面执行的destory方法是一个什么方法。

方法destory

  public void destroy() {
    executor.shutdown();
  }

很显然，就是关闭接口和执行器并释放资源。回到源码中来，方法最后就将两个全局变量设置为空。

小结

本周的Hive源码解析任务完成了。至此，我们把plan文件夹下的mapper文件夹中的所有文件代码都解析完毕了。从下一周开始，我们将开始解析同样是在plan文件夹下的ptf文件夹中的文件代码，希望能够学习到更多的知识。

【软件工程实践】Hive研究-Blog8

大数据系统相关栏目本月热门文章