经过一番搜索，我设法仅从 REST API 提交了一个应用程序。这不是一个有据可查的过程，所以我将其发布在这里。
注意：如果您希望将请求的内容与客户端发送的请求进行比较，请使用调试断点检查客户端使用的应用程序上下文。
打开类org.apache.hadoop.yarn.client.api.impl.YarnClientImpl，进入方法submitApplication(ApplicationSubmissionContext appContext)。
首先，要将spark.deploy.yarn.Client 替换为 REST API 请求，解决方案必须确保配置中提到的所有文件在 HDFS 上都可用。
稍后，它需要编写并上传一个名为__spark_conf__.zip 的额外文件。
第一步
检查来自SparkConf 的文件（Client 的第二个参数）：“AllJars”标签中提到的文件，“mainJarPath 中提到的文件”，以及“FilesList”中提到的文件。
对于每个文件，检查它是否存在于 HDFS 中，如果不存在 - 从本地机器上传。对于每个文件，从 HDFS 获取其FileStatus。
聚合资源列表，即包含这6个属性的每个文件的属性映射：
- size = getLen()
- 时间戳 = getModificationTime()
- type=文件
- visibility=公开
另外两个属性：键和资源。
- allJars 列表中的文件：键是 spark_libs/{{filename}}，资源是文件名。
- FilesList 中的文件：键是“localEntry”标签，资源是“hdfsPath”标签。
- mainJarPath 中的文件：key 为“app.jar”，resource 为文件名。
第 2 步
创建__spark_conf__.zip 文件。您可以直接在 hdfs 中创建它，在通常为{{HDFS_base_folder}}/user/{{username}}/.sparkStaging/{{application_id}}/__spark_conf__.zip 的暂存路径中。
这个存档文件包含两个文件和一个空目录：一个文件__spark_hadoop_conf__.xml（重命名为core-site.xml），另一个文件名为__spark_conf__.properties，这是一个稍微修改的版本
sparkConf 部分的配置。
要创建__spark_conf__.properties，您需要从 "sparkConf"->"org$apache$spark$SparkConf$$settings" 读取 JSON 映射，并将每一行从 JSON 格式转换"spark.safemine.addcontrol.driverMemory": "5120M"
到 spark.safemine.addcontrol.driverMemory=5120M
在文件底部添加 6 个新行：
- spark.yarn.cache.confArchive={{您将在 sparkStaging 中上传
__spark_conf__.zip 的位置}}
- spark.yarn.cache.visibilities={{所有文件的可见性，以逗号分隔 - 基本上是“PUBLIC,PUBLIC, ... ,PUBLIC”}}
- spark.yarn.cache.timestamps={{所有文件的时间戳，逗号分隔}}
- spark.yarn.cache.types={{所有文件类型，逗号分隔 - 基本上是 "FILE,FILE, ... ,FILE"}}
- spark.yarn.cache.filenames={{所有文件名和键，记录为resource#key，逗号分隔}}
- spark.yarn.cache.sizes={{文件的所有大小，逗号分隔}}
确保您按各自的顺序编译 5 个聚合行。我使用了这段代码：
String confArchive = "spark.yarn.cache.confArchive="+hdfs+"/user/"+userName+"/.sparkStaging/"+applicationId+"/__spark_conf__.zip";
String filenames = "spark.yarn.cache.filenames=";
String sizes = "spark.yarn.cache.sizes=";
String timestamps = "spark.yarn.cache.timestamps=";
String types = "spark.yarn.cache.types=";
String visibilities = "spark.yarn.cache.visibilities=";
for (Map<String,String> localResource:localResources) {
filenames+=localResource.get("resource")+"#"+localResource.get("key")+",";
sizes+=localResource.get("size")+",";
timestamps+=localResource.get("timestamp")+",";
types+=localResource.get("type")+",";
visibilities+=localResource.get("visibility")+",";
}
properties+=confArchive+"\n";
properties+=filenames.substring(0,filenames.length()-1)+"\n";
properties+=sizes.substring(0,sizes.length()-1)+"\n";
properties+=timestamps.substring(0,timestamps.length()-1)+"\n";
properties+=types.substring(0,types.length()-1)+"\n";
properties+=visibilities.substring(0,visibilities.length()-1)+"\n";
__spark_hadoop_conf__.xml 文件是 core-site.xml 的简单重命名，使用它们创建的文件夹命名为 __hadoop_conf__ 并留空。
您可以像这样直接将文件保存到 hdfs：
private void generateSparkConfInHdfs(String applicationId, String userName, String sparkConfProperties, String sparkHadoopConf) throws IOException {
String path = hdfs+"/user/"+userName+"/.sparkStaging/"+applicationId+"/__spark_conf__.zip";
Path hdfsPath = new Path(path);
ZipOutputStream os = new ZipOutputStream(getHdfs().create(hdfsPath));
os.putNextEntry(new ZipEntry("__hadoop_conf__/"));
os.putNextEntry(new ZipEntry("__spark_conf__.properties"));
os.write(sparkConfProperties.getBytes(),0,sparkConfProperties.getBytes().length);
os.putNextEntry(new ZipEntry("__spark_hadoop_conf__.xml"));
os.write(sparkHadoopConf.getBytes(),0,sparkHadoopConf.getBytes().length);
os.close();
}
创建完文件后，将其添加到具有以下规范的资源列表中：
- size = getLen()
- 时间戳 = getModificationTime()
- 类型=存档
- 可见性 = 私人
- 键 =
__spark_conf__
- resource 是暂存目录（通常是
{{HDFS_base_folder}}/user/{{username}}/.sparkStaging/{{application_id}}/__spark_conf__.zip）。
查看完整资源列表，并使用我们在 {{}} 占位符中收集的值，为每个资源创建一个具有这种结构的 XML/JSON：
<entry>
<key>{{key}}</key>
<value>
<resource>{{resource}}</resource>
<size>{{size}}</size>
<timestamp>{{timestamp}}</timestamp>
<type>{{type}}</type>
<visibility>{{visibility}}</visibility>
</value>
</entry>
累积的字符串将是您的localResources XML 段，如下所示。
第三步
生成 Java 命令。您需要从 SparkConfig 中提取一些元素：
- driverMemory - 来自
sparkConf 中的相同属性
- extraJavaOptions = 来自属性集合中的
spark.driver.extraJavaOptions
- mainClass - 来自
sparkConf 中的相同属性
- argstr - 收集除 --class 之外的所有
ClientArgs。
包含元素的结果命令是：
String command = "$JAVA_HOME/bin/java -server -Xmx"+driverMemory+" -Djava.io.tmpdir=$PWD/tmp "+extraJavaOptions+" -Dspark.yarn.app.container.log.dir=<LOG_DIR> "
+ "org.apache.spark.deploy.yarn.ApplicationMaster --class "+mainClass+" "+argstr+" "
+ "--properties-file $PWD/__spark_conf__/__spark_conf__.properties 1> <LOG_DIR>/stdout 2> <LOG_DIR>/stderr";
第四步
编译请求 XML。
注意：我的实现需要 AM 容器上的标签，因此添加了 am-container-node-label-expression。这并不适用于所有情况。
sparkConf 到 REST 请求的映射是（这里显示为 XML，也支持 JSON 实现）：
<application-submission-context>
<application-id>"+applicationId+"</application-id>
<application-name>"+appName+"</application-name>
<queue>default</queue>
<priority>0</priority>
<am-container-spec>
<local-resources>+localResources+</local-resources>
<environment>
<entry>
<key>SPARK_YARN_STAGING_DIR</key>
<value>"+hdfs+"/user/"+userName+"/.sparkStaging/"+applicationId+"</value>
</entry>
<entry>
<key>CLASSPATH</key>
<value>$PWD:$PWD/__spark_conf__:$PWD/__spark_libs__/*:/spark-non-hdfs-storage/spark-assembly-2.3.0-hadoop2.7/*:%HADOOP_CONF_DIR%:%HADOOP_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>"+userName+"</value>
</entry>
</environment>
<commands>
<command>"+command+"</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>"+sparkYarnTag+"</tag>
</application-tags>
<am-container-node-label-expression>appMngr</am-container-node-label-expression>
<log-aggregation-context/>
<attempt-failures-validity-interval>1</attempt-failures-validity-interval>
<reservation-id/>
</application-submission-context>
第 5 步：
通过 REST http PUT 提交申请：
private void submitApplication (String body, String userName) throws SMSparkManagerException {
HttpClient client = HttpClientBuilder.create().build();
HttpPost request = new HttpPost(uri+"?user.name="+userName);
try {
request.setEntity(new StringEntity(body, ContentType.APPLICATION_XML));
HttpResponse response = client.execute(request);
if (response.getStatusLine().getStatusCode()!=202) {
throw new SMSparkManagerException("The application could not be submitted to Yarn, response http code "+response.getStatusLine().getStatusCode());
}
} catch (UnsupportedEncodingException e) {
logger.error("The application Could not be submitted due to UnsupportedEncodingException in the provided body: "+body, e );
throw new SMSparkManagerException("Error in submitting application to yarn");
} catch (ClientProtocolException e) {
logger.error("The application Could not be submitted due to ClientProtocolException", e);
throw new SMSparkManagerException("Error in submitting application to yarn");
} catch (IOException e) {
logger.error("The application Could not be submitted due to IOException", e);
throw new SMSparkManagerException("Error in submitting application to yarn");
}
}