通过REST提交EMR纱线应用程序

Question

我在AWS中有一个带有YARN的Hadoop集群，我向其提交了spark应用程序。我通过REST请求进行工作，提交本文中指定的XML：YARNRESTneneneba API。它适用于常规集群。

我目前正在做一个POC，用于处理EMR集群，而不是通常的POC，在那里我使用现有的REST命令，并简单地通过SSH与EMR的内部YARN通信，如下所述：内部EMR服务的Web访问。它适用于大多数REST命令，如POST http://<rm http address:port>/ws/v1/cluster/apps/new-application，但当我提交新应用程序时，它会立即失败，并报告找不到ApplicationMaster。

日志类型：stderr

日志上传时间：2019年2月3日星期日17:18:35+00000

日志长度：88

错误：无法找到或加载主类org.apache.spark.deploy.yarn.ApplicationMaster

我怀疑这是以某种方式连接到类路径的，当我向REST提交应用程序节点添加带有所有jar的EMR FS位置的类路径标志(/usr/lib/spark/jars/*)时，它找到了ApplicationMaster，但随后在核心实例中找不到jars，出现了以下奇怪的错误日志：

日志类型：stderr

日志上传时间：2019年1月31日星期四15:11:21+000

日志长度：89

错误：无法找到或加载主类/usr/lib.spark.jars.datanucleus-core-3.2.10.jar

最不寻常的是它试图描述找不到的jar的方式，而不是类。经过进一步的调查，我发现了原因：当Java命令被发送到Core实例时，它将类路径解析为它的三个文件：java -server -classpath /usr/lib/spark/jars/datanucleus-api-jdo-3.2.6.jar /usr/lib/spark/jars/datanucleus-core-3.2.10.jar /usr/lib/spark/jars/datanucleus-rdbms-3.2.9.jar ...，因此尝试执行"/usr/lib/spark/jars/datanucleus-Core-3.2.10.jar"，就好像它是可运行的一样。问题是，如果我试图将类路径更改为更具体的类路径，或者如果我试图删除它，那么应用程序会因为找不到ApplicationMaster而再次失败。

我发送给YARN的REST请求是：

<?xml version="1.0" encoding="UTF-8" standalone="yes"?> 
<application-submission-context> 
<application-id>application_1549270910165_0001</application-id> 
<application-name> .... some name .....</application-name> 
<queue>default</queue> 
<priority>0</priority> 
<am-container-spec> 
<local-resources> 
<entry>
....... MANY MANY RESOURCES ......
</value>
</entry>
</local-resources> 
<environment> 
<entry> 
<key>SPARK_YARN_STAGING_DIR</key> 
<value>..... staging directory in our HDFS ..... </value> 
</entry> 
<entry> 
<key>CLASSPATH</key> 
<value>$PWD:$PWD/__spark_conf__:$PWD/__spark_libs__/*:/usr/lib/spark/jars/*:/usr/lib/spark/yarn/lib/*:%HADOOP_CONF_DIR%:%HAOOP_COMMON_HOME%/share/hadoop/common/*:%HADOOP_COMMON_HOME%/share/hadoop/common/lib/*:%HADOOP_HDFS_HOME%/share/hadoop/hdfs/*:%HADOOP_HDFS_HOME%/share/hadoop/hdfs/lib/*:%HADOOP_YARN_HOME%/share/hadoop/yarn/*:%HADOOP_YARN_HOME%/share/hadoop/yarn/lib/*:%HADOOP_MAPRED_HOME%/share/hadoop/mapreduce/*:%HADOOP_MAPRED_HOME%/share/hadoop/mapreduce/lib/*:$PWD/__spark_conf__/__hadoop_conf__</value> 
</entry>
<entry>
<key>SPARK_USER</key>
<value>... user name ....</value>
</entry>
</environment>
<commands> 
<command>command=$JAVA_HOME/bin/java -classpath '/usr/lib/spark/jars/*' -server -Xmx5120M -Djava.io.tmpdir=$PWD/tmp '-XX:hashCode=0' '-Dlog4j.configuration=log4j-client.cfg' '-Dhdp.version=2.8.4' -Dspark.yarn.app.container.log.dir=&lt;LOG_DIR&gt; org.apache.spark.deploy.yarn.ApplicationMaster ... some jar and arguments ....  --properties-file $PWD/__spark_conf__/__spark_conf__.properties 1&gt; &lt;LOG_DIR&gt;/stdout 2&gt; &lt;LOG_DIR&gt;/stderr</command> 
</commands> 
</am-container-spec> 
<unmanaged-AM>false</unmanaged-AM> 
<max-app-attempts>1</max-app-attempts> 
<resource> 
<memory>5632</memory> 
<vCores>1</vCores> 
</resource> 
<application-type>SPARK</application-type> 
<keep-containers-across-application-attempts>false</keep-containers-across-application-attempts> 
<application-tags> 
<tag>.... product tag .....</tag> 
</application-tags> 
<log-aggregation-context/> 
<attempt-failures-validity-interval>1</attempt-failures-validity-interval> 
<reservation-id/> 
</application-submission-context>

如果有线索，我将不胜感激。

Answer 1

经过长时间的搜索，我发现应用程序无法加载类org.apache.spark.deploy.yarn.ApplicationMaster的原因是因为这不是EMR核心实例使用的ApplicationMaster版本——它使用org.apache.hadoop.yarn.applications.distributedshell.ApplicationMaster，这需要输入中的CLASSPATH段包含/usr/lib/hadoop-yarn/*。我更改了REST请求的输入XML中的两个参数，它成功地启动了。我仍然需要为EMR实现配置正确的CLASSPATH，以使应用程序成功完成，但这个问题的主要挑战已经解决。

更新：最终我决定在EMR中添加一个步骤并使用参数实际上有一种更容易的方法来处理它。我在maven依赖项中添加了EMR AWS Java SDK:

<dependency>
<groupId>com.amazonaws</groupId>
<artifactId>aws-java-sdk-emr</artifactId>
<version>1.11.486</version>
</dependency>

并添加了以下代码：

AddJobFlowStepsResult result = emr.addJobFlowSteps(new AddJobFlowStepsRequest()
.withJobFlowId(clusterId)
.withSteps(new StepConfig()
.withName(name)
.withActionOnFailure(ActionOnFailure.CONTINUE)
.withHadoopJarStep(new HadoopJarStepConfig()
.withJar("command-runner.jar")
.withArgs(stepargs))));

stepargs取自我最初的REST请求，包括要使用的jar和文件——简单地使用spark-submit：

List<String> stepargs = new ArrayList<String>();
stepargs.add("spark-submit");
stepargs.add("--class");
stepargs.add(mainClass);
stepargs.add("--deploy-mode");
stepargs.add("cluster");
stepargs.add("--master");
stepargs.add("yarn");
stepargs.add("--files");
stepargs.add(files);
stepargs.add("--jars");
stepargs.add(jars);
stepargs.add("--properties-file");
stepargs.add(confFileName);
stepargs.add(jar);
Iterator<String> itr = args.iterator();
while (itr.hasNext()) {
String arg = itr.next();
if (arg.equals("--arg")) {
stepargs.add(itr.next());
}
}