Spark任务提交至YARN运行的3种方式与Mapreduce提交任务的几种形式

一: Spark任务提交至YARN运行的3种方式

Spark作为新一代计算平台的闪亮明星，在我们的大数据平台中具有非常重要的作用，SQL查询、流计算和机器学习等场景都能见到它的身影，可以说平台应用的数据处理、计算和挖掘等场景都可以使用Spark进行开发。在默认的情况下，如果想向Spark提交计算任务，通常会使用Spark提供的Spark-Submit脚本来提交含有业务逻辑的jar文件程序。这种方式虽然简单，但有悖于服务化的设计理念，所以需要为Spark提供一套任务管理的RESTful服务。
在大数据平台中，Spark是以Spark on YARN的方式运行的，在这种模式下，整个集群的资源调度是由YARN统一控制的，Spark只是作为运行在YARN上的一个应用客户端而存在。本文将介绍提交Spark任务至YARN运行的3种方式。如下图所示：

1. 第一种方式：使用Spark-Submit脚本提交

Spark本身提供了Spark-Submit脚本用于提交任务，可以借助Java的Process-Builder调用脚本，将其包装成RESTful服务。

 ./spark-submit --class com.learn.spark.SimpleApp --master yarn --deploy-mode client --driver-memory 2g --executor-memory 2g --executor-cores 3 ../spark-demo.jar

2. 第二种方式：使用Spark Client提交

 除了Spark-Submit脚本之外，Spark还提供了一套Java客户端接口用于提交任务。在使用这套接口之后，程序就可以去掉对Spark-Submit脚本的依赖，这样一来提交任务的服务程序就可以运行在应用服务器之上，使得以远程的方式向集群提交任务成为可能。提交方式是已JAVA API编程的方式提交，这种方式不需要使用命令行，直接可以在IDEA中点击Run 运行包含Job的Main类就行。Spark 提供了以SparkLanuncher 作为唯一入口的API来实现。这种方式很方便（试想如果某个任务需要重复执行，但是又不会写linux 脚本怎么搞？我想到的是以JAVA API的方式提交Job, 还可以和Spring整合，让应用在tomcat中运行），官网的示例：[官网示例链接](http://spark.apache.org/docs/latest/api/java/index.html?org/apache/spark/launcher/package-summary.html)

根据官网的示例，通过JAVA API编程的方式提交有两种方式：
第一种是调用SparkLanuncher实例的startApplication方法，但是这种方式在所有配置都正确的情况下使用运行都会失败的，原因是startApplication方法会调用LauncherServer启动一个进程与集群交互，这个操作貌似是异步的，所以可能结果是main主线程结束了这个进程都没有起起来，导致运行失败。解决办法是调用new SparkLanuncher().startApplication后需要让主线程休眠一定的时间后者是使用下面的例子：

package com.learn.spark; import org.apache.spark.launcher.SparkAppHandle;
import org.apache.spark.launcher.SparkLauncher; import java.io.IOException;
import java.util.HashMap;
import java.util.concurrent.CountDownLatch; public class LanuncherAppV { public static void main(String[] args) throws IOException, InterruptedException { HashMap env = new HashMap(); //这两个属性必须设置 env.put("HADOOP_CONF_DIR", "/usr/local/hadoop/etc/overriterHaoopConf"); env.put("JAVA_HOME", "/usr/local/java/jdk1.8.0_151"); //可以不设置 //env.put("YARN_CONF_DIR",""); CountDownLatch countDownLatch = new CountDownLatch(1); //这里调用setJavaHome()方法后，JAVA_HOME is not set 错误依然存在 SparkAppHandle handle = new SparkLauncher(env) .setSparkHome("/usr/local/spark") .setAppResource("/usr/local/spark/spark-demo.jar") .setMainClass("com.learn.spark.SimpleApp") .setMaster("yarn") .setDeployMode("cluster") .setConf("spark.app.id", "11222") .setConf("spark.driver.memory", "2g") .setConf("spark.akka.frameSize", "200") .setConf("spark.executor.memory", "1g") .setConf("spark.executor.instances", "32") .setConf("spark.executor.cores", "3") .setConf("spark.default.parallelism", "10") .setConf("spark.driver.allowMultipleContexts", "true") .setVerbose(true).startApplication(new SparkAppHandle.Listener() { //这里监听任务状态，当任务结束时（不管是什么原因结束）,isFinal（）方法会返回true,否则返回false @Override public void stateChanged(SparkAppHandle sparkAppHandle) { if (sparkAppHandle.getState().isFinal()) { countDownLatch.countDown(); } System.out.println("state:" + sparkAppHandle.getState().toString()); } @Override public void infoChanged(SparkAppHandle sparkAppHandle) { System.out.println("Info:" + sparkAppHandle.getState().toString()); } }); System.out.println("The task is executing, please wait ...."); //线程等待任务结束 countDownLatch.await(); System.out.println("The task is finished!"); }
}

注意：如果部署模式是cluster,但是代码中有标准输出的话将看不到，需要把结果写到HDFS中，如果是client模式则可以看到输出。

第二种方式是：通过SparkLanuncher.lanunch()方法获取一个进程，然后调用进程的process.waitFor()方法等待线程返回结果，但是使用这种方式需要自己管理运行过程中的输出信息，比较麻烦，好处是一切都在掌握之中，即获取的输出信息和通过命令提交的方式一样，很详细，实现如下：

package com.learn.spark; import org.apache.spark.launcher.SparkAppHandle;
import org.apache.spark.launcher.SparkLauncher; import java.io.IOException;
import java.util.HashMap; public class LauncherApp { public static void main(String[] args) throws IOException, InterruptedException { HashMap env = new HashMap(); //这两个属性必须设置 env.put("HADOOP_CONF_DIR","/usr/local/hadoop/etc/overriterHaoopConf"); env.put("JAVA_HOME","/usr/local/java/jdk1.8.0_151"); //env.put("YARN_CONF_DIR",""); SparkLauncher handle = new SparkLauncher(env) .setSparkHome("/usr/local/spark") .setAppResource("/usr/local/spark/spark-demo.jar") .setMainClass("com.learn.spark.SimpleApp") .setMaster("yarn") .setDeployMode("cluster") .setConf("spark.app.id", "11222") .setConf("spark.driver.memory", "2g") .setConf("spark.akka.frameSize", "200") .setConf("spark.executor.memory", "1g") .setConf("spark.executor.instances", "32") .setConf("spark.executor.cores", "3") .setConf("spark.default.parallelism", "10") .setConf("spark.driver.allowMultipleContexts","true") .setVerbose(true); Process process =handle.launch(); InputStreamReaderRunnable inputStreamReaderRunnable = new InputStreamReaderRunnable(process.getInputStream(), "input"); Thread inputThread = new Thread(inputStreamReaderRunnable, "LogStreamReader input"); inputThread.start(); InputStreamReaderRunnable errorStreamReaderRunnable = new InputStreamReaderRunnable(process.getErrorStream(), "error"); Thread errorThread = new Thread(errorStreamReaderRunnable, "LogStreamReader error"); errorThread.start(); System.out.println("Waiting for finish..."); int exitCode = process.waitFor(); System.out.println("Finished! Exit code:" + exitCode); }
}

package com.learn.spark; import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader; public class InputStreamReaderRunnable implements Runnable { private BufferedReader reader; private String name; public InputStreamReaderRunnable(InputStream is, String name) { this.reader = new BufferedReader(new InputStreamReader(is)); this.name = name; } public void run() {System.out.println("InputStream " + name + ":"); try { String line = reader.readLine(); while (line != null) { System.out.println(line); line = reader.readLine(); } reader.close(); } catch (IOException e) { e.printStackTrace(); } }
}

3. 第三种方式：使用YARN RESTful API提交

除了Spark-Submit和SparkClient两种方法之外，还可以通过YARN提供的RESTful API向其提交Spark任务。但是这种方法十分复杂，并不推荐使用。

Post请求示例： * http:///ws/v1/cluster/apps

官网介绍地址
既然是api，推荐使用postman或者idea，本文使用postman

0.准备MR的jar包并上传到hdfs上

hdfs dfs -put xifan227-1.0-SNAPSHOT.jar /cpos/test/

3.1.生成新的applicationId

请求类型：POST
URL：http://rm-http-address:port/ws/v1/cluster/apps/new-application

3.2.查看MR的jar包信息(查看modificationTime)

请求类型：GET
URL：http://nm-http-address:port/webhdfs/v1/cpos/test/xifan227-1.0-SNAPSHOT.jar?op=GETFILESTATUS

3.3.提交应用 (注意timestamp)

Cluster Applications API(Submit Application)
请求类型：POST
URL：http://rm-http-address:port/ws/v1/cluster/apps
Body：

{"application-id":"application_1586340793234_0125","application-name":"AaronTest","queue":"default","priority":-2,"am-container-spec":{"local-resources":{"entry":[{"key":"xifan227-1.0-SNAPSHOT","value":{"resource":"hdfs://test96.com:8020/cpos/test/xifan227-1.0-SNAPSHOT.jar","type":"FILE","visibility":"APPLICATION","size": 30301,"timestamp": 1586400852288}}]},"commands":{"command":"/opt/modules/hadoop-3.2.1/bin/yarn jar xifan227-1.0-SNAPSHOT mr.WordcountMapreduce 1><LOG_DIR>/WC.stdout 2><LOG_DIR>/WC.stderr"}},"max-app-attempts":1,"resource":{"memory":1024, "vCores":2},"application-type":"MAPREDUCE"
}

4.结果展示

二: Mapreduce任务提交的几种方式

1．问题来源

为什么我们在windows下用编程集成工具或者linux下用编程集成工具来run我们编写的mr程序的时候是在本地jvm中运行，而我们在linux上用命令行：hadoop jar 提交我们的jar文件的时候就提交到集群去运行了呢？
具体原因是：如果是RunJar中包含了和RM通信的rpc客户端就是把jar包提交到集群运行，如果RunJar中包含了和本地机器通讯的客户端的话，就是把jar包提交到本地的jvm中运行。这些都是根据配置文件来配置的。
Configurationconf = new Configuration().是由这个来决定的。如果你conf里面设定了mapreduce.framework.name的value为yarn的话，它就会初始化一个持有和yarn通信的rpc客户端，如果没有设置，则会初始化一个和本地通信的客户端。

2.提交模式分类

本地提交模式
① 在windows或者linux上访问的本地文件系统上的文件，生成的结果也是在本地文件系统上面
这种模式的启动模式是本地启动，并没有将程序上传到集群里面去。
② 在windows或者linux上访问的文件是hdfs上的文件，生成的结果也是输出到hdfs上面。
这种模式虽然读取的是hdfs上的文件，但是也没有将程序上传到集群去执行，最终还是一种本地启动模式。

这个staging的资源文件是在本地文件系统上面，而非是在hdfs上面，所以还是一种本地启动模式。
集群提交模式
③ 把程序打成jar包，上传到服务器用hadoop命令提交 Hadoop jar jar的名字 jar的main方法路径，这样是一种集群提交模式
④ 在linux的eclipse中运行程序，当把配置文件拷进到程序的classpath下面之后(主要是mapred-site.xml文件和yarn-site.xml) ，这也是一种集群提交模式。
先读到mapred-site.xml中的mapreduce.framework.name值，初始化一个和RM通信的rpc客户端，然后将任务提交到RM去。

此时执行程序的时候会报错

说是找不到mapper类，因为在eclipse中没有打jar包，而且没有指定jar在哪，所以找不到。
可以在配置文件中设置这个属性，让程序执行的时候可以找到jar文件。

conf.set("mapreduce.framework.name","wordcount.jar")

然后将jar包打在工程工程目录下就行了。
⑤ 在windows的eclipse中直接运行main方法，配置项和linux中利用eclipse运行main方法差不多，也是一种集群提交模式，但是由于平台不兼容，需要做很多的设置修改，网上也有许多资料介绍怎么修改，可以自行查阅。但是不推荐这种模式。