YARN DistributedShell源码分析与修改

共计 14410 个字符，预计需要花费 37 分钟才能阅读完成。

YARN DistributedShell 源码分析与修改

1 概述
2 YARN DistributedShell 不能满足当前需求
- 2.1 功能需求
- 2.2 YARN DistributedShell 对需求的支持情况
- 2.3 需要对 YARN DistributedShell 进行的修改
3 YARN DistributedShell 源码获取
4 YARN DistributedShell 源码分析及修改

4.1 Client 类
- 4.1.1 Client 源码逻辑
- 4.1.2 对 Client 源码的修改
4.2 ApplicationMaster 类
- 4.2.1 ApplicationMaster 源码逻辑
- 4.2.2 对 ApplicationMaster 源码的修改
4.3 DSConstants 类
4.4 Log4jPropertyHelper 类

1 概述

Hadoop YARN 项目自带一个非常简单的应用程序编程实例 –DistributedShell。DistributedShell 是一个构建在 YARN 之上的 non-MapReduce 应用示例。它的主要功能是在 Hadoop 集群中的多个节点，并行执行用户提供的 shell 命令或 shell 脚本（将用户提交的一串 shell 命令或者一个 shell 脚本，由 ApplicationMaster 控制，分配到不同的 container 中执行)。

2 YARN DistributedShell 不能满足当前需求

2.1 功能需求

我所参与的项目通过融合 Hive、MapReduce、Spark、Kafka 等大数据开源组件，搭建了一个数据分析平台。
平台需要新增一个功能：

在集群中选取一个节点，执行用户提交的 jar 包。
该功能需要与平台已有的基于 Hive、MR、Spark 实现的业务以及 YARN 相融合。
简而言之，经分析与调研，我们需要基于 YARN 的 DistributedShell 实现该功能。

该功能需要实现：

单机执行用户自己提交的 jar 包
用户提交的 jar 包会有其他 jar 包的依赖
用户提交的 jar 包只能选取一个节点运行
用户提交的 jar 包需要有缓存数据的目录

2.2 YARN DistributedShell 对需求的支持情况

YARN 的 DistributedShell 功能为：

支持执行用户提供的 shell 命令或脚本
执行节点数可以通过参数 num_containers 设置，默认值为 1
不支持 jar 包的执行
更不支持依赖包的提交
不支持 jar 包缓存目录的设置

2.3 需要对 YARN DistributedShell 进行的修改

增加支持执行 jar 包功能
增加支持缓存目录设置功能
删除执行节点数设置功能，不允许用户设置执行节点数，将执行节点数保证值为 1

3 YARN DistributedShell 源码获取

YARN DistributedShell 源码可以在 GitHub 上 apache/hadoop 获取，hadoop repository 中 DistributedShell 的源代码路径为:
hadoop/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-applications/hadoop-yarn-applications-distributedshell/src/main/java/org/apache/hadoop/yarn/applications/distributedshell/
这里修改的是 2.6.0 版本源码。

4 YARN DistributedShell 源码分析及修改

YARN DistributedShell 包含 4 个 java Class：

DistributedShell
    ├── Client.java
    ├── ApplicationMaster.java
    ├── DSConstants.java
    ├── Log4jPropertyHelper.java

Client：客户端提交 application
ApplicationMaster：注册 AM，申请分配 container，启动 container
DSConstants：Client 类和 ApplicationMaster 类中的常量定义
Log4jPropertyHelper：加载 Log4j 配置

4.1 Client 类

4.1.1 Client 源码逻辑

Client 类是 DistributedShell 应用提交到 YARN 的客户端。Client 将启动 application master，然后 application master 启动多个 containers 用于运行 shell 命令或脚本。Client 运行逻辑为：

使用 ApplicationClientProtocol 协议连接 ResourceManager（也叫 ApplicationsMaster 或 ASM），获取一个新的 ApplicationId。（ApplicationClientProtocol 提供给 Client 一个获取集群信息的方式）
在一个 job 提交过程中，Client 首先创建一个 ApplicationSubmissionContext。ApplicationSubmissionContext 定义了 application 的详细信息，例如：ApplicationId、application name、application 分配的优先级、application 分配的队列。另外，ApplicationSubmissionContext 还定义了一个 Container，该 Container 用于启动 ApplicationMaster。
在 ContainerLaunchContext 中需要初始化启动 ApplicationMaster 的资源：
- 运行 ApplicationMaster 的 container 的资源
- jars（例：AppMaster.jar）、配置文件(例：log4j.properties)
- 运行环境（例：hadoop 特定的类路径、java classpath）
- 启动 ApplicationMaster 的命令
Client 使用 ApplicationSubmissionContext 提交 application 到 ResourceManager，并通过按周期向 ResourceManager 请求 ApplicationReport，完成对 applicatoin 的监控。
如果 application 运行时间超过 timeout 的限制（默认为 600000 毫秒，可通过 -timeout 进行设置），client 将发送 KillApplicationRequest 到 ResourceManager，将 application 杀死。

具体代码如下(基于 YARN2.6.0)：

Cilent 的入口 main 方法：

publicstaticvoidmain(String[] args) {boolean result = false;
        try {DshellClient client = new DshellClient();
            LOG.info("Initializing Client");
            try {boolean doRun = client.init(args);
                if (!doRun) {System.exit(0);
                }
            } catch (IllegalArgumentException e) {System.err.println(e.getLocalizedMessage());
                client.printUsage();
                System.exit(-1);
            }
            result = client.run();} catch (Throwable t) {LOG.fatal("Error running Client", t);
            System.exit(1);
        }
        if (result) {LOG.info("Application completed successfully");
            System.exit(0);
        }
        LOG.error("Application failed to complete successfully");
        System.exit(2);
    }

main 方法：

输入参数为用户 CLI 的执行命令，例如：hadoop jar hadoop-yarn-applications-distributedshell-2.0.5-alpha.jar org.apache.hadoop.yarn.applications.distributedshell.Client -jar hadoop-yarn-applications-distributedshell-2.0.5-alpha.jar -shell_command '/bin/date' -num_containers 10, 该命令提交的任务为：启动 10 个 container，每个都执行 date 命令。
main 方法将运行 init 方法，如果 init 方法返回 true 则运行 run 方法。
init 方法解析用户提交的命令，解析用户命令中的参数值。
run 方法将完成 Client 源码逻辑中描述的功能。

4.1.2 对 Client 源码的修改

在原有 YARN DistributedShell 的基础上做的修改如下：

在 CLI 为用户增加了 container_files 和container_archives两个参数
- container_files指定用户要执行的 jar 包的依赖包，多个依赖包以逗号分隔
- container_archives指定用户执行的 jar 包的缓存目录，多个目录以逗号分隔
删除 num_containers 参数
- 不允许用户设置 container 的个数，使用默认值 1

对 Client 源码修改如下：

变量
- 增加变量用于保存 container_files 和container_archives两个参数的值

// 增加两个变量，保存 container_files、container_archives 的参数值↓↓↓↓↓↓↓
private String[] containerJarPaths = new String[0];
private String[] containerArchivePaths = new String[0];
// ↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑

Client 构造方法
- 删除 num_containers 参数的初试��，增加 container_files 和container_archives两个参数
- 修改构造方法的 ApplicationMaster 类

// 删除 num_containers 项，不允许用户设置 containers 个数，containers 个数默认为 1 ↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓
//opts.addOption("num_containers", true, "No. of containers on which the shell command needs to be executed");
// ↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑
// 添加 container_files、container_archives 的描述↓↓↓↓↓↓↓↓↓↓↓↓↓↓
this.opts.addOption("container_files", true,"The files that containers will run .  Separated by comma");
this.opts.addOption("container_archives", true,"The archives that containers will unzip.  Separated by comma");
// ↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑

publicDshellClient(Configuration conf) throws Exception {// 修改构造方法的 ApplicationMaster 类↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓
        this("org.apache.hadoop.yarn.applications.distributedshell.DshellApplicationMaster",conf);
        // ↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑
    }

init 方法
- 增加 container_files 和container_archives两个参数的解析

// 初始化选项 container_files、container_archives↓↓↓↓↓↓↓
this.opts.addOption("container_files", true,"The files that containers will run .  Separated by comma");
this.opts.addOption("container_archives", true,"The archives that containers will unzip.  Separated by comma");
// ↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑

run 方法
- 上传 container_files 和container_archives两个参数指定的依赖包和缓存目录至 HDFS

 // 上传 container_files 指定的 jar 包到 HDFS ↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓
if (this.containerJarPaths.length != 0)
    for (int i = 0; i < this.containerJarPaths.length; i++) {String hdfsJarLocation = "";
        String[] jarNameSplit = this.containerJarPaths[i].split("/");
        String jarName = jarNameSplit[(jarNameSplit.length - 1)];

        long hdfsJarLen = 0L;
        long hdfsJarTimestamp = 0L;
        if (!this.containerJarPaths[i].isEmpty()) {Path jarSrc = new Path(this.containerJarPaths[i]);
            String jarPathSuffix = this.appName + "/" + appId.toString() +
                    "/" + jarName;
            Path jarDst = new Path(fs.getHomeDirectory(), jarPathSuffix);
            fs.copyFromLocalFile(false, true, jarSrc, jarDst);
            hdfsJarLocation = jarDst.toUri().toString();
            FileStatus jarFileStatus = fs.getFileStatus(jarDst);
            hdfsJarLen = jarFileStatus.getLen();
            hdfsJarTimestamp = jarFileStatus.getModificationTime();
            env.put(DshellDSConstants.DISTRIBUTEDJARLOCATION + i,
                    hdfsJarLocation);
            env.put(DshellDSConstants.DISTRIBUTEDJARTIMESTAMP + i,
                    Long.toString(hdfsJarTimestamp));
            env.put(DshellDSConstants.DISTRIBUTEDJARLEN + i,
                    Long.toString(hdfsJarLen));
        }
    }
// ↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑
// 上传 container_archives 到 HDFS↓↓↓↓↓↓↓↓↓↓↓↓↓↓
long hdfsArchiveLen;
String archivePathSuffix;
Path archiveDst;
FileStatus archiveFileStatus;
if (this.containerArchivePaths.length != 0) {for (int i = 0; i < this.containerArchivePaths.length; i++) {String hdfsArchiveLocation = "";
        String[] archiveNameSplit = this.containerArchivePaths[i].split("/");
        String archiveName = archiveNameSplit[(archiveNameSplit.length - 1)];
        hdfsArchiveLen = 0L;
        long hdfsArchiveTimestamp = 0L;
        if (!this.containerArchivePaths[i].isEmpty()) {Path archiveSrc = new Path(this.containerArchivePaths[i]);
            archivePathSuffix = this.appName + "/" + appId.toString() +
                    "/" + archiveName;
            archiveDst = new Path(fs.getHomeDirectory(),
                    archivePathSuffix);
            fs.copyFromLocalFile(false, true, archiveSrc, archiveDst);
            hdfsArchiveLocation = archiveDst.toUri().toString();
            archiveFileStatus = fs.getFileStatus(archiveDst);
            hdfsArchiveLen = archiveFileStatus.getLen();
            hdfsArchiveTimestamp = archiveFileStatus
                    .getModificationTime();
            env.put(DshellDSConstants.DISTRIBUTEDARCHIVELOCATION + i,
                    hdfsArchiveLocation);
            env.put(DshellDSConstants.DISTRIBUTEDARCHIVETIMESTAMP + i,
                    Long.toString(hdfsArchiveTimestamp));
            env.put(DshellDSConstants.DISTRIBUTEDARCHIVELEN + i,
                    Long.toString(hdfsArchiveLen));
        }
    }
}
// ↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑

4.2 ApplicationMaster 类

4.2.1 ApplicationMaster 源码逻辑

一个 ApplicationMaster 将在启动一个或过个 container，在 container 上执行 shell 命令或脚本。ApplicationMaster 运行逻辑为：

ResourceManager 启动一个 container 用于运行 ApplicationMaster。
ApplicationMaster 连接 ResourceManager，向 ResourceManager 注册自己。
- 向 ResourceManager 注册的信息有：
  - ApplicationMaster 的 ip:port
  - ApplicationMaster 所在主机的 hostname
  - ApplicationMaster 的 tracking url。客户端可以用 tracking url 来跟踪任务的状态和历史记录。
- 需要注意的是：在 DistributedShell 中，不需要初注册 tracking url 和 appMasterHost:appMasterRpcPort，只需要设置 hostname。
ApplicationMaster 会按照设定的时间间隔向 ResourceManager 发送心跳。ResourceManager 的 ApplicationMasterService 每次收到 ApplicationMaster 的心跳信息后，会同时在 AMLivelinessMonitor 更新其最近一次发送心跳的时间。
ApplicationMaster 通过 ContainerRequest 方法向 ResourceManager 发送请求，申请相应数目的 container。在发送申请 container 请求前，需要初始化 Request，需要初始化的参数有：
- Priority：请求的优先级
- capability：当前支持 CPU 和 Memory
- nodes：申请的 container 所在的 host（如果不需要指定，则设为 null）
- racks：申请的 container 所在的 rack（如果不需要指定，则设为 null）
ResourceManager 返回 ApplicationMaster 的申请的 containers 信息，根据 container 的状态 -containerStatus，更新已申请成功和还未申请的 container 数目。
申请成功的 container，ApplicationMaster 则通过 ContainerLaunchContext 初始化 container 的启动信息。初始化 container 后启动 container。需要初始化的信息有：
- Container id
- 执行资源（Shell 脚本或命令、处理的数据）
- 运行环境
- 运行命令
container 运行期间，ApplicationMaster 对 container 进行监控。
job 运行结束，ApplicationMaster 发送 FinishApplicationMasterRequest 请求给 ResourceManager，完成 ApplicationMaster 的注销。

具体代码如下(基于 YARN2.6.0)：

ApplicationMaster 的入口 main 方法：

publicstaticvoidmain(String[] args) {boolean result = false;
       try {DshellApplicationMaster appMaster = new DshellApplicationMaster();
           LOG.info("Initializing ApplicationMaster");
           boolean doRun = appMaster.init(args);
           if (!doRun) {System.exit(0);
           }
           appMaster.run();
           result = appMaster.finish();} catch (Throwable t) {LOG.fatal("Error running ApplicationMaster", t);
           LogManager.shutdown();
           ExitUtil.terminate(1, t);
       }
       if (result) {LOG.info("Application Master completed successfully. exiting");
           System.exit(0);
       } else {LOG.info("Application Master failed. exiting");
           System.exit(2);
       }
   }

main 方法：

输入参数为 Client 提交的执行命令。
init 方法完成对执行命令的解析，获取执行命令中参数指定的值。
run 方法完成 ApplicationMaster 的启动、注册、containers 的申请、分配、监控等功能的启动。
- run 方法中建立了与 ResourceManager 通信的 Handle-AMRMClientAsync，其中的 CallbackHandler 是由 RMCallbackHandler 类实现的。
  - RMCallbackHandler 类中实现了 containers 的申请、分配等方法。
  - containers 的分配方法 onContainersAllocated 中通过 LaunchContainerRunnable 类中 run 方法完成 container 的启动。
finish 方法完成 container 的停止、ApplicationMaster 的注销。

4.2.2 对 ApplicationMaster 源码的修改

在原有 YARN DistributedShell 的基础上做的修改如下：

在 ApplicationMaster 初试化时，增加对 container_files 和container_archives两个参数指定值的支持。即：初始化 container_files 和container_archives指定的运行资源在 HDFS 上的信息。
在 container 运行时，从 HDFS 上加载 container_files 和container_archives指定的资源。

对 ApplicationMaster 源码修改如下：

变量
- 增加变量，用于保存 container_files 和container_archives指定的运行资源在 HDFS 上的信息。

// 增加 container_files、container_archives 选项值变量 ↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓
private ArrayList<DshellFile> scistorJars = new ArrayList();
private ArrayList<DshellArchive> scistorArchives = new ArrayList();
// ↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑

ApplicationMaster 的 init 方法
- 初始化 container_files 和container_archives两个参数指定值信息。

// 遍历 envs，把所有的 jars、archivers 的 HDFS 路径，时间戳，LEN 全部保存到 jarPaths 对象数组中 ↓↓↓↓↓↓↓↓↓↓
for (String key : envs.keySet()) {if (key.contains(DshellDSConstants.DISTRIBUTEDJARLOCATION)) {DshellFile scistorJar = new DshellFile();
        scistorJar.setJarPath((String) envs.get(key));
        String num = key
                .split(DshellDSConstants.DISTRIBUTEDJARLOCATION)[1];
        scistorJar.setTimestamp(Long.valueOf(Long.parseLong((String) envs
                        .get(DshellDSConstants.DISTRIBUTEDJARTIMESTAMP + num))));
        scistorJar.setSize(Long.valueOf(Long.parseLong((String) envs
                        .get(DshellDSConstants.DISTRIBUTEDJARLEN + num))));
        this.scistorJars.add(scistorJar);
    }
}

for (String key : envs.keySet()) {if (key.contains(DshellDSConstants.DISTRIBUTEDARCHIVELOCATION)) {DshellArchive scistorArchive = new DshellArchive();
        scistorArchive.setArchivePath((String) envs.get(key));
        String num = key
                .split(DshellDSConstants.DISTRIBUTEDARCHIVELOCATION)[1];
        scistorArchive.setTimestamp(Long.valueOf(Long.parseLong((String) envs
                        .get(DshellDSConstants.DISTRIBUTEDARCHIVETIMESTAMP +
                                num))));
        scistorArchive.setSize(Long.valueOf(Long.parseLong((String) envs
                        .get(DshellDSConstants.DISTRIBUTEDARCHIVELEN + num))));
        this.scistorArchives.add(scistorArchive);
    }
}
// ↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑

LaunchContainerRunnable 的 run 方法（container 线程的 run 方法）
- 从 HDFS 上加载 container_files 和container_archives指定的资源。

// 把 HDFS 中的 jar、archive 加载到 container 的 LocalResources，也就是从 HDFS 分发到 container 节点的过程 ↓↓↓↓↓↓↓↓↓↓↓↓↓
for (DshellFile perJar : DshellApplicationMaster.this.scistorJars) {LocalResource jarRsrc = (LocalResource) Records.newRecord(LocalResource.class);
    jarRsrc.setType(LocalResourceType.FILE);
    jarRsrc.setVisibility(LocalResourceVisibility.APPLICATION);
    try {jarRsrc.setResource(ConverterUtils.getYarnUrlFromURI(new URI(perJar.getJarPath()
                        .toString())));
    } catch (URISyntaxException e1) {DshellApplicationMaster.LOG.error("Error when trying to use JAR path specified in env, path=" +
                perJar.getJarPath(), e1);
        DshellApplicationMaster.this.numCompletedContainers.incrementAndGet();
        DshellApplicationMaster.this.numFailedContainers.incrementAndGet();
        return;
    }
    jarRsrc.setTimestamp(perJar.getTimestamp().longValue());
    jarRsrc.setSize(perJar.getSize().longValue());
    String[] tmp = perJar.getJarPath().split("/");
    localResources.put(tmp[(tmp.length - 1)], jarRsrc);
}
String[] tmp;
for (DshellArchive perArchive : DshellApplicationMaster.this.scistorArchives) {
    LocalResource archiveRsrc =
            (LocalResource) Records.newRecord(LocalResource.class);
    archiveRsrc.setType(LocalResourceType.ARCHIVE);
    archiveRsrc.setVisibility(LocalResourceVisibility.APPLICATION);
    try {archiveRsrc.setResource(ConverterUtils.getYarnUrlFromURI(new URI(perArchive
                        .getArchivePath().toString())));
    } catch (URISyntaxException e1) {DshellApplicationMaster.LOG.error("Error when trying to use ARCHIVE path specified in env, path=" +
                        perArchive.getArchivePath(),
                e1);
        DshellApplicationMaster.this.numCompletedContainers.incrementAndGet();
        DshellApplicationMaster.this.numFailedContainers.incrementAndGet();
        return;
    }
    archiveRsrc.setTimestamp(perArchive.getTimestamp().longValue());
    archiveRsrc.setSize(perArchive.getSize().longValue());
    tmp = perArchive.getArchivePath().split("/");
    String[] tmptmp = tmp[(tmp.length - 1)].split("[.]");
    localResources.put(tmptmp[0], archiveRsrc);
}
// ↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑

4.3 DSConstants 类

DSConstants 类中是在 Client 和 ApplicationMaster 中的常量，对 DSConstants 类的修改为：增加了 container_files、container_archives 相关常量。修改代码如下：

// 增加 container_files、container_archives 相关常量 ↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓
public static final String DISTRIBUTEDJARLOCATION = "DISTRIBUTEDJARLOCATION";
public static final String DISTRIBUTEDJARTIMESTAMP = "DISTRIBUTEDJARTIMESTAMP";
public static final String DISTRIBUTEDJARLEN = "DISTRIBUTEDJARLEN";

public static final String DISTRIBUTEDARCHIVELOCATION = "DISTRIBUTEDARCHIVELOCATION";
public static final String DISTRIBUTEDARCHIVETIMESTAMP = "DISTRIBUTEDARCHIVETIMESTAMP";
public static final String DISTRIBUTEDARCHIVELEN = "DISTRIBUTEDARCHIVELEN";
// ↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑