Skip to content

Client API

Jump to bottom
Parker Zhou edited this page Oct 20, 2020 · 1 revision

ClientAPI提供提交应用到yarn上的基本API，还提供对提交的应用进行查询、变更、扩缩容等操作。Cli和Web提供的用户接口本质上都是基于ClientAPI的。

ClientAPI的代码位于marayarn-client下，本节详细介绍ClientAPI的使用。

基本调用

提交一个应用运行的代码十分简单，下面代码摘自marayarn-cli，展示了如何调用ClientAPI提交一个应用:

ClientArguments clientArguments = SubmitOptions.toClientArguments(commandLine);
try(Client client = new Client()) {
    ApplicationReport report = client.launch(clientArguments);
    logger.info("application id: {}", report.getApplicationId());
}

首选需要创建ClientArguments，然后实例化Client类，调用launch，传入ClientArguments即可。

ClientArguments

ClientArguments封装所有跟提交应用有关的参数，下面对每个参数做详细说明

applicationId

应用在yarn上的applicationId。提交应用的时候不需要这个参数，但是状态查询、变更、删除、扩缩容等操作需要。

applicationMasterJar

提交应用必需。ApplicationMaster的jar包，对应marayarn-am编译打包出来的jar包。marayarn本质上就是开了一个ApplicationMaster叫MaraApplicationMaster，它负责运行、维护应用进程。

Client需要上传MaraApplicationMaster，所以必须知道jar包的位置。可以用file://, hdfs://, http://, ftp://表示的URI，Client会自动下载并上传到HDFS上。

applicationName

提交应用必需。可以自定义的应用名。

applicationMasterClass

ApplicationMasterClass的主类。除非定制化ApplicationMaster，否则默认为com.eoi.marayarn.MaraApplicationMaster。

hadoopConfDir

HADOOP配置文件所在本路径，Client必须知道这个路径，才能定位yarn和hdfs。如果不指定的话，会依次尝试检查相关环境变量和CLassPath

  1. $HADOOP_CONF_DIR
  2. $HADOOP_HOME/conf
  3. $HADOOP_HOME/etc/conf
  4. java ClassPath

queue

设置yarn资源池的队列名

cpu/memory

设置实例的资源配额。vcore, 默认为1; memory, 单位MB, 默认为512

instances

设置初次提交的实例数量。表示希望同时启动多少个实例

user

设置proxy-user。这个参数与principal和keytab不能同时存在。这个行为与spark一致

aMEnvironments

内部保留，设置ApplicationMaster的环境变量

executorEnvironments

设置待执行的程序的环境变量

artifacts

设置Executor需要的物料，Artifact的集合，Client会上传这些到hdfs，可以用file://, hdfs://, http://, ftp://表示的URI。

Client会将文件传输到当前hdfs环境，如果目标URI跟当前hdfs是同一个环境，则会跳过上传步骤。其他格式的URI都会将文件上传到`$HOME/.stage/$APPLICATION_ID。

支持使用归档文件(tar.gz, zip格式)，yarn会自动解压。URI最后的fragment部分会作为resource的key告知yarn，yarn在解压缩的时候，会以这个key作为外层目录，例如：

file:///Users/pchou/Downloads/logstash-7.3.0.tar.gz#dir
http://host:9999/jaxjobjarlib/spark/logstash-7.3.0.tar.gz#dir

都会上传logstash-7.3.0.tar.gz，解压后container的当前目录下有dir/logstash/...

command

提交任务必须。执行程序的命令行，例如：

dir/logstash/bin/logstash -e 'input { stdin { } } output { stdout {} }'

principal/keytab

kerberos认证必需的principal/keytab

constraints

类似marathon的约束。约束描述了一种进程分配在哪个物理节点上运行的策略，基本格式为：

field,operator[,value]
  • field可以是node,rack
  • operator支持CLUSTER,LIKE,UNLIKE,IS,UNIQUE,GROUP_BY
  • value一般是对应字符串或正则。对于CLUSTER, value可以不填; 对于UNIQUE/GROUP_BY, value不需要填

下面是一些constraints及其含义:

  • node,CLUSTER: 随机选择一个node，所有实例部署到这个node上
  • node,LIKE,vm319[0-9]: 实例分布在满足node名称为vm319[0-9]的节点上
  • rack,IS,/rack1: 实例分布在/rack1的节点上
  • rack,UNIQUE: 实例在每种rack中只能有唯一的实例
  • node,UNIQUE: 实例在所有node上最多只有一个实例
  • node,CLUSTER,vm3195: 同IS, 实例全部分布在vm3195上
  • node,GROUP_BY: 实例按照node分散部署，相当于负载均衡的部署

示例

下面的代码可以提交一个logstash，并运行：

ClientArguments arguments = new ClientArguments();
arguments.setApplicationMasterJar("file:///Users/pchou/Projects/java/marayarn/marayarn-am/target/marayarn-am-1.0-SNAPSHOT-jar-with-dependencies.jar");
arguments.setApplicationName("marayarn_test2");
List<Artifact> artifacts = new ArrayList<>();
Artifact tar = new Artifact()
        .setLocalPath("http://192.168.31.55:9999/jaxjobjarlib/spark/logstash-7.3.0-eoi.tar.gz#dir")
        .setType(LocalResourceType.ARCHIVE);
artifacts.add(tar);
arguments.setArtifacts(artifacts);
arguments.setCommand("dir/logstash/bin/logstash -e 'input { stdin { } } output { stdout {} }'");
Client client = new Client();
ApplicationReport report = client.launch(arguments);
System.out.println(report.getTrackingUrl());

下面代码提交一个bash脚本到带有kerberos认证的yarn上，并指定了约束：6个实例都必须在运行在/rack01机架上

ClientArguments arguments = new ClientArguments();
arguments.setApplicationMasterJar("file://" + getProjectRoot() + "/marayarn-am/target/" + AM_JAR_NAME);
arguments.setApplicationName("test_no_constraints");
arguments.setHadoopConfDir(getProjectRoot() + "/hadoop-kerberos");
arguments.setPrincipal("mara@ALANWANG.COM");
arguments.setKeytab("file://" + getProjectRoot() + "/hadoop-kerberos/mara.keytab");
arguments.setCommand("while true; do date; sleep 5; done");
arguments.setInstances(6);
arguments.setConstraints("rack,CLUSTER,/rack01");
Client client = new Client();
ApplicationReport report = client.launch(arguments);
System.out.println(report.getTrackingUrl());

Clone this wiki locally