Drill-on-YARN之源码解析-白红宇

1. 概要

前面介绍了如何把Drill部署在YARN上，然后通过Drill-on-YARN客户端，你可以启动、停止、调整、清零命令操作Drill。但是在这么命令背后，到底是如何执行的呢，下面会对Drill-on-YARN的源码进行详细的解析，重点解析启动过程，其他命令简单介绍。

说明：下面涉及到的代码，以drill 1.14.0为准，并且为了减少篇幅，进行了删减。

2. Drill-on-YARN start

2.1 drill-on-yarn.sh

通过查看drill-on-yarn.sh脚本，很容易发现最终执行的java类是CLIENT_CMD="$JAVA $VM_OPTS -cp $CP org.apache.drill.yarn.client.DrillOnYarn ${args[@]}"。 org.apache.drill.yarn.client.DrillOnYarn便是启动Drill-on-YARN的入口。我们可以总览一下这个类：

public class DrillOnYarn {  public static void main(String argv[]) {    BasicConfigurator.configure();    ClientContext.init();    run(argv);  }  public static void run(String argv[]) {    ClientContext context = ClientContext.instance();    CommandLineOptions opts = new CommandLineOptions();    if (!opts.parse(argv)) {      opts.usage();      context.exit(-1);    }    if (opts.getCommand() == null) {      opts.usage();      context.exit(-1);    }    try {      DrillOnYarnConfig.load().setClientPaths();    } catch (DoyConfigException e) {      ClientContext.err.println(e.getMessage());      context.exit(-1);    }    ClientCommand cmd;    switch (opts.getCommand()) {    case UPLOAD:      cmd = new StartCommand(true, false);      break;    case START:      cmd = new StartCommand(true, true);      break;    case DESCRIBE:      cmd = new PrintConfigCommand();      break;    case STATUS:      cmd = new StatusCommand();      break;    case STOP:      cmd = new StopCommand();      break;    case CLEAN:      cmd = new CleanCommand();      break;    case RESIZE:      cmd = new ResizeCommand();      break;    default:      cmd = new HelpCommand();    }    cmd.setOpts(opts);    try {      cmd.run();    } catch (ClientException e) {      displayError(opts, e);      context.exit(1);    }  }}

可以看到入口main方法，其中最关键的便是run方法，包含了很多的命令，我们重点看start命令，代码如下：

public void run() throws ClientException {    checkExistingApp();    dryRun = opts.dryRun;    config = DrillOnYarnConfig.config();    FileUploader uploader = upload();    if (launch) {      launch(uploader);    }}

概括的来说，它主要包含以下流程：

检查application是否已经存在，如果已经存在，便不允许启动，否则执行启动操作（此处检查的application是YARN的application，启动成功会将YARN的applicationId写入本地磁盘的一个文件，通过此文件来检查）。

上传Drill二方包和site目录下的内容至DFS上，其中site目录下的内容会被打包为site.tar.gz

public void run() throws ClientException {  setup();  uploadDrillArchive();  if (hasSiteDir()) {    uploadSite();  }}

启动ApplicationMaster，主要流程为：

创建YARN客户端，并启动

// AMRunner#connectToYarnprivate void connectToYarn() {    System.out.print("Loading YARN Config...");    client = new YarnRMClient();    System.out.println(" Loaded.");}

创建ApplicationMaster

// AMRunner#createAppprivate void createApp() throws ClientException {    try {      appResponse = client.createAppMaster();    } catch (YarnClientException e) {      throw new ClientException("Failed to allocate Drill application master",          e);    }    appId = appResponse.getApplicationId();    System.out.println("Application ID: " + appId.toString());}

设置ApplicationMaster上下文，包括：Heap memory、Class Path、启动的命令（dirll-am.sh）、启动am容器使用的资源（memory、vCores、disks）

校验资源，主要是ApplicationMaster使用资源是否超过了YARN的设置

提交ApplicationMaster

private void launchApp(AppSpec master) throws ClientException {    try {      client.submitAppMaster(master);    } catch (YarnClientException e) {      throw new ClientException("Failed to start Drill application master", e);    }}

等待启动，并打印启动日志

将ApplicationMaster的appid写入文件（在第1步，检测Application是否存在，就是使用这个文件）

ApplicationMaster启动后，会向RM申请资源，启动Drillbits，下面详细介绍ApplicationMaster启动后的操作

2.2 drill-am.sh

通过查看drill-am.sh脚本，很容易发现最终执行的java类是AMCMD="$JAVA $AM_JAVA_OPTS ${args[@]} -cp $CP org.apache.drill.yarn.appMaster.DrillApplicationMaster"。org.apache.drill.yarn.appMaster.DrillApplicationMaste表示ApplicationMaster执行的入口，下面总览一下这个类：

public class DrillApplicationMaster {  public static void main(String[] args) {    LOG.trace("Drill Application Master starting.");    try {      DrillOnYarnConfig.load().setAmDrillHome();    } catch (DoyConfigException e) {      System.err.println(e.getMessage());      System.exit(-1);    }    Dispatcher dispatcher;    try {      dispatcher = (new DrillControllerFactory()).build();    } catch (ControllerFactoryException e) {      LOG.error("Setup failed, exiting: " + e.getMessage(), e);      System.exit(-1);      return;    }    try {      if (!dispatcher.start()) {        return;      }    } catch (Throwable e) {      LOG.error("Fatal error, exiting: " + e.getMessage(), e);      System.exit(-1);    }    WebServer webServer = new WebServer(dispatcher);    try {      webServer.start();    } catch (Exception e) {      LOG.error("Web server setup failed, exiting: " + e.getMessage(), e);      System.exit(-1);    }    try {      dispatcher.run();    } catch (Throwable e) {      LOG.error("Fatal error, exiting: " + e.getMessage(), e);      System.exit(-1);    } finally {      try {        webServer.close();      } catch (Exception e) {      }    }  }}

概况的来说，它主要包含以下流程：

加载Drill-on-YARN的配置，并设置AM的DirllHome，比如/home/admin/tmp2/hadoop/nm-local-dir/usercache/admin/appcache/application_1534698866098_0022/container_1534698866098_0022_01_000001/drill/apache-drill-1.14.0

构造Dispatcher，Dispatcher用于分配YARN、timer、ZooKeeper事件给给集群控制器，它是轻量级多线程的，用于响应RM、NM、timer线程的事件，对于某一个事件，它是连续的，所以需要同步，但是不同类型的事件不需要同步。整个的构造流程如下：

准备资源，包括：drill二方包、site压缩包的目录

private Map
        
          prepareResources() {    ...    drillArchivePath = drillConfig.getDrillArchiveDfsPath();    siteArchivePath = drillConfig.getSiteArchiveDfsPath();    ...}

定义任务启动的规格（TaskSpec），包括：运行时环境、YARN container的规格、dirllbit的规格

private TaskSpec buildDrillTaskSpec(Map
        
          resources) throws DoyConfigException {    ...    ContainerRequestSpec containerSpec = new ContainerRequestSpec();    containerSpec.memoryMb = config.getInt(DrillOnYarnConfig.DRILLBIT_MEMORY);    ...    LaunchSpec drillbitSpec = new LaunchSpec();    ...    TaskSpec taskSpec = new TaskSpec();    taskSpec.name = "Drillbit";    taskSpec.containerSpec = containerSpec;    taskSpec.launchSpec = drillbitSpec;}

设置Dispatcher的控制器：实现类为ClusterControllerImpl，它主要通过状态来控制Drill集群、调整整个集群的任务（Drill启动、停止等任务）、处理container的回调
```
public void setYarn(AMYarnFacade yarn) throws YarnFacadeException {    this.yarn = yarn;    controller = new ClusterControllerImpl(yarn);}
```

为控制器注册Scheduler，比如DrillbitScheduler，此外Scheduler配置来源于之前drill-on-yarn.conf

cluster: [    {      name: "drill-group1"      type: "basic"      count: 1    }]

...ClusterDef.ClusterGroup pool = ClusterDef.getCluster(config, 0);Scheduler testGroup = new DrillbitScheduler(pool.getName(), taskSpec,pool.getCount(), requestTimeoutSecs, maxExtraNodes);dispatcher.getController().registerScheduler(testGroup);...

创建ZooKeeper集群协调器

String zkConnect = config.getString(DrillOnYarnConfig.ZK_CONNECT);String zkRoot = config.getString(DrillOnYarnConfig.ZK_ROOT);String clusterId = config.getString(DrillOnYarnConfig.CLUSTER_ID);

启动Dispatcher，主要启动AMRMClientAsync、NMClientAsync、YarnClient

...yarn.start(new ResourceCallback(), new NodeCallback());String url = trackingUrl.replace("
      
       ", Integer.toString(httpPort));if (DrillOnYarnConfig.config().getBoolean(DrillOnYarnConfig.HTTP_ENABLE_SSL)) {  url = url.replace("http:", "https:");}yarn.register(url);controller.started();...

...resourceMgr = AMRMClientAsync.createAMRMClientAsync(pollPeriodMs, resourceCallback);resourceMgr.init(conf);resourceMgr.start();...nodeMgr = NMClientAsync.createNMClientAsync(nodeCallback);nodeMgr.init(conf);nodeMgr.start();...client = YarnClient.createYarnClient();client.init(conf);client.start();...

启动dirll运维界面

WebServer webServer = new WebServer(dispatcher);webServer.start();

运行Dispatcher，主要是启动一个线程，此线程会不断的轮询当前的任务队列中的任务情况，比如启动、停止、resize等类型的任务，然后执行相应的动作，拿启动来说

添加一个启动任务，然后放入pendingTask队列中

if (state == State.LIVE) {  adjustTasks(curTime);  requestContainers();}

向RM请求container：创建一个ContainerRequest

ContainerRequest request = containerSpec.makeRequest();resourceMgr.addContainerRequest(containerSpec.makeRequest());return request;

ResourceCallback监听container分配，然后启动container

private class ResourceCallback implements AMRMClientAsync.CallbackHandler {    @Override    public void onContainersAllocated(List
        
          containers) {      controller.containersAllocated(containers);    }}

public void containerAllocated(EventContext context, Container container) {  Task task = context.task;  LOG.info(task.getLabel() + " - Received container: "      + DoYUtil.describeContainer(container));  context.group.dequeueAllocatingTask(task);  // No matter what happens below, we don't want to ask for this  // container again. The RM async API is a bit bizarre in this  // regard: it will keep asking for container over and over until  // we tell it to stop.  context.yarn.removeContainerRequest(task.containerRequest);  // The container is need both in the normal and in the cancellation  // path, so set it here.  task.container = container;  if (task.cancelled) {    context.yarn.releaseContainer(container);    taskStartFailed(context, Disposition.CANCELLED);    return;  }  task.error = null;  task.completionStatus = null;  transition(context, LAUNCHING);  // The pool that manages this task wants to know that we have  // a container. The task manager may want to do some task-  // specific setup.  context.group.containerAllocated(context.task);  context.getTaskManager().allocated(context);  // Go ahead and launch a task in the container using the launch  // specification provided by the task group (pool).  try {    context.yarn.launchContainer(container, task.getLaunchSpec());    task.launchTime = System.currentTimeMillis();  } catch (YarnFacadeException e) {    LOG.error("Container launch failed: " + task.getContainerId(), e);    // This may not be the right response. RM may still think    // we have the container if the above is a local failure.    task.error = e;    context.group.containerReleased(task);    task.container = null;    taskStartFailed(context, Disposition.LAUNCH_FAILED);  }}

NodeCallback监听container启动

public class NodeCallback implements NMClientAsync.CallbackHandler {    @Override    public void onStartContainerError(ContainerId containerId, Throwable t) {      controller.taskStartFailed(containerId, t);    }    @Override    public void onContainerStarted(ContainerId containerId, Map
        
          allServiceResponse) {      controller.containerStarted(containerId);    }    @Override    public void onContainerStatusReceived(ContainerId containerId, ContainerStatus containerStatus) {    }    @Override    public void onGetContainerStatusError(ContainerId containerId, Throwable t) {    }    @Override    public void onStopContainerError(ContainerId containerId, Throwable t) {      controller.stopTaskFailed(containerId, t);    }    @Override    public void onContainerStopped(ContainerId containerId) {      controller.containerStopped(containerId);    }}

2.3 fail over

Drill-on-YARN除了提供start、stop、resize功能外，还提供了fail over功能，当前某个drillbit挂掉后，Drill-on-YARN会尝试再次启动drillbit，目前重试的次数为2。此外，如果一个drillbit所在的节点频繁挂掉，会被列入黑名单。

我们可以通过手动kill drillbit来模拟drillbit挂掉的情况，然后等待一会儿，可以看到，drillbit进程重新启动了。下面我们看看，代码的执行流程

drillbit挂掉，container结束

private class ResourceCallback implements AMRMClientAsync.CallbackHandler {    @Override    public void onContainersCompleted(List
    
      statuses) {      controller.containersCompleted(statuses);    }}

retry task：重新将这个task加入pendingTasks，然后轮询的线程检测到pendingTasks不为空，执行启动操作

protected void taskTerminated(EventContext context) {    Task task = context.task;    context.getTaskManager().completed(context);    context.group.containerReleased(task);    assert task.completionStatus != null;    // container结束的状态不是0，说明不是正常结束    if (task.completionStatus.getExitStatus() == 0) {      taskEnded(context, Disposition.COMPLETED);      context.group.taskEnded(context.task);    } else {      taskEnded(context, Disposition.RUN_FAILED);      retryTask(context);    }}

private void retryTask(EventContext context) {    Task task = context.task;    assert task.state == END;    if (!context.controller.isLive() || !task.retryable()) {      context.group.taskEnded(task);      return;    }    if (task.tryCount > task.taskGroup.getMaxRetries()) {      LOG.error(task.getLabel() + " - Too many retries: " + task.tryCount);      task.disposition = Disposition.TOO_MANY_RETRIES;      context.group.taskEnded(task);      return;    }    LOG.info(task.getLabel() + " - Retrying task, try " + task.tryCount);    context.group.taskRetried(task);    task.reset();    transition(context, START);    context.group.enqueuePendingRequest(task);}

3. 停止

除了前面详情介绍的start命令外，Drill-on-YARN也提供了stop命令，其中stop分两种：

强制停止：直接调用yarn客户端的killApplication api yarnClient.killApplication(appId);

优雅停止：先清理所有的任务，包括pending、running的，然后调用yarn的api杀死容器，关闭controller，然后通知am运行结束

...for (Task task : getStartingTasks()) {  context.setTask(task);  context.getState().cancel(context);}for (Task task : getActiveTasks()) {  context.setTask(task);  context.getState().cancel(context);}...

...context.yarn.killContainer(task.container);...

public void run() throws YarnFacadeException {    ...    boolean success = controller.waitForCompletion();    ...    ...    finish(success, null);    ...  }

public boolean waitForCompletion() {    start();    synchronized (completionMutex) {      try {        completionMutex.wait();      } catch (InterruptedException e) {              }    }    return succeeded();}

public void finish(boolean succeeded, String msg) throws YarnFacadeException {    nodeMgr.stop();    String appMsg = "Drill Cluster Shut-Down";    FinalApplicationStatus status = FinalApplicationStatus.SUCCEEDED;    if (!succeeded) {      appMsg = "Drill Cluster Fatal Error - check logs";      status = FinalApplicationStatus.FAILED;    }    if (msg != null) {      appMsg = msg;    }    try {      resourceMgr.unregisterApplicationMaster(status, appMsg, "");    } catch (YarnException | IOException e) {      throw new YarnFacadeException("Deregister AM failed", e);    }    resourceMgr.stop();  }

4. resize

resize流程为：调整quantity（保留多少个container），之后轮询线程会根据quantity，调整任务，执行resize操作

public int resize(int level) {    int limit = quantity + state.getController().getFreeNodeCount() +maxExtraNodes;    return super.resize( Math.min( limit, level ) );}

5. 总结

总的来说，Drill-on-YARN分为两大模块，drill-on-yarn.sh和drill-am.sh。drill-on-yarn.sh用于启动ApplicationMaster，drill-am.sh用于向ResourceManager申请资源并启动Drill集群。其中Drill的启动、停止、缩容、扩容，都被封装为一个任务，在执行这些命令时，会构建一个任务，放入任务队列中。有一个线程会一直轮询此队列，根据队列中的任务执行不同的操作，从而达到启动、停止、缩容、扩容Drill集群的功能。此外，相比独立部署，Drill-on-YARN提供的failover功能强化了Drill的稳定性。