| 注册
请输入搜索内容

热门搜索

Java Linux MySQL PHP JavaScript Hibernate jQuery Nginx
jopen
10年前发布

spark用程序提交任务到yarn

因为spark文档中只介绍了两种用脚本提交到yarn的例子,并没有介绍如何通过程序提交yarn,但是我们的需求需要这样。网上很难找到例子,经过几天摸索,终于用程序提交到yarn成功,下面总结一下。

先介绍官网提交的例子,我用的是spark 0.9.0 hadoop2.2.0

一.使用脚本提交

1.使用spark脚本提交到yarn,首先需要将spark所在的主机和hadoop集群之间hosts相互配置(也就是把spark主机的ip和主机名配置到hadoop所有节点的/etc/hosts里面,再把集群所有节点的ip和主机名配置到spark所在主机的/etc/hosts里面)。

2.然后需要把hadoop目录etc/hadoop下面的*-sit.xml复制到${SPARK_HOME}的conf下面.

3.确保hadoop集群配置了 HADOOP_CONF_DIR or YARN_CONF_DIR 

1.yarn-standalone方式提交到yarn

${SPARK_HOME}下面执行:

SPARK_JAR=./assembly/target/scala-2.10.4/spark-assembly-0.9.0-incubating-hadoop2.2.0.jar \      ./bin/spark-class org.apache.spark.deploy.yarn.Client \        --jar ./examples/target/scala-2.10/spark-examples_2.10-assembly-0.9.0-incubating.jar \        --class org.apache.spark.examples.SparkPi \        --args yarn-standalone \        --num-workers 3 \        --master-memory 2g \        --worker-memory 2g \        --worker-cores 1



2. yarn-client 方式提交到yarn


${SPARK_HOME}下面执行:

SPARK_JAR=./assembly/target/scala-2.10.4/spark-assembly-0.9.0-incubating-hadoop2.2.0.jar \  SPARK_YARN_APP_JAR=examples/target/scala-2.10/spark-examples_2.10-assembly-0.9.0-incubating.jar \  ./bin/run-example org.apache.spark.examples.SparkPi yarn-client



二、使用程序提交


1.必须使用linux主机提交任务,使用windows提交到linux hadoop集群会报


org.apache.hadoop.util.Shell$ExitCodeException: /bin/bash: 第 0 行: fg: 无任务控制


错误。hadoop2.2.0不支持windows提交到linux hadoop集群,网上搜索发现这是hadoop的bug。

2.提交任务的主机和hadoop集群主机名需要在hosts相互配置。

3.因为使用程序提交是使用yarn-client方式,所以必须像上面脚本那样设置环境变量SPARK_JAR 和 SPARK_YARN_APP_JAR

比如我的设置为向提交任务主机~/.bashrc里面添加:


export SPARK_JAR=file:///home/ndyc/software/sparkTest/lib/spark-assembly-0.9.0-incubating-hadoop2.2.0.jar    export SPARK_YARN_APP_JAR=file:///home/ndyc/software/sparkTest/ndspark-0.0.1.jar


如果不加file:// spark会任务是hdfs,会在hdfs里面找,file://表明是本地文件。


其中SPARK_JAR是${SPARK_HOME}/assembly/target/scala-2.10.4/spark-assembly-0.9.0-incubating-hadoop2.2.0.jar

SPARK_YARN_APP_JAR是自己程序打的jar包,包含自己的测试程序。

4.程序中加入hadoop、yarn、依赖。

注意,如果引入了hbase依赖,需要这样配置


<dependency>              <groupId>org.apache.hbase</groupId>              <artifactId>hbase-thrift</artifactId>              <version>${hbase.version}</version>              <exclusions>                  <exclusion>                      <groupId>org.apache.hadoop</groupId>                      <artifactId>hadoop-mapreduce-client-jobclient</artifactId>                  </exclusion>                  <exclusion>                      <groupId>org.apache.hadoop</groupId>                      <artifactId>hadoop-client</artifactId>                  </exclusion>              </exclusions>          </dependency>

然后再加入


<dependency>              <groupId>org.ow2.asm</groupId>              <artifactId>asm-all</artifactId>              <version>4.0</version>          </dependency>


否则会报错:



IncompatibleClassChangeError has interface org.objectweb.asm.ClassVisitor as super class



异常是因为Hbase jar hadoop-mapreduce-client-jobclient.jar里面使用到了asm3.1 而spark需要的是asm-all-4.0.jar

5. hadoop conf下的*-site.xml需要复制到提交主机的classpath下,或者说maven项目resources下面。

6.编写程序

代码示例:


package com.sdyc.ndspark.sys;    import org.apache.spark.SparkConf;  import org.apache.spark.api.java.JavaPairRDD;  import org.apache.spark.api.java.JavaRDD;  import org.apache.spark.api.java.JavaSparkContext;  import org.apache.spark.api.java.function.Function2;  import org.apache.spark.api.java.function.PairFunction;  import scala.Tuple2;    import java.util.ArrayList;  import java.util.List;    /**   * Created with IntelliJ IDEA.   * User: zarchary   * Date: 14-1-19   * Time: 下午6:23   * To change this template use File | Settings | File Templates.   */  public class ListTest {        public static void main(String[] args) throws Exception {          SparkConf sparkConf = new SparkConf();          sparkConf.setAppName("listTest");          //使用yarn模式提交          sparkConf.setMaster("yarn-client");            JavaSparkContext sc = new JavaSparkContext(sparkConf);            List<String> listA = new ArrayList<String>();            listA.add("a");          listA.add("a");          listA.add("b");          listA.add("b");          listA.add("b");          listA.add("c");          listA.add("d");            JavaRDD<String> letterA = sc.parallelize(listA);            JavaPairRDD<String, Integer> letterB = letterA.map(new PairFunction<String, String, Integer>() {              @Override              public Tuple2<String, Integer> call(String s) throws Exception {                  return new Tuple2<String, Integer>(s, 1);              }          });            letterB = letterB.reduceByKey(new Function2<Integer, Integer, Integer>() {              public Integer call(Integer i1, Integer i2) {                  return i1 + i2;              }          });            //颠倒顺序          JavaPairRDD<Integer, String> letterC = letterB.map(new PairFunction<Tuple2<String, Integer>, Integer, String>() {              @Override              public Tuple2<Integer, String> call(Tuple2<String, Integer> stringIntegerTuple2) throws Exception {                  return new Tuple2<Integer, String>(stringIntegerTuple2._2, stringIntegerTuple2._1);              }          });            JavaPairRDD<Integer, List<String>> letterD = letterC.groupByKey();  //        //false说明是降序          JavaPairRDD<Integer, List<String>> letterE = letterD.sortByKey(false);            System.out.println("========" + letterE.collect());            System.exit(0);      }  }



代码中master设置为yar-client表明了是使用提交到yarn.


关于spark需要依赖的jar的配置可以参考我的博客spark安装和远程调用。

以上弄完之后就可以运行程序了。

运行后会看到yarn的ui界面出现:


正在执行的过程中会发现hadoop yarn 有的nodemanage会有下面这个进程:


13247 org.apache.spark.deploy.yarn.WorkerLauncher


这是spark的工作进程。


如果接收到异常为:

WARN YarnClientClusterScheduler: Initial job has not accepted any resources; check your cluster UI to ensure that workers are registered and have sufficient memory

 出现这个错误是因为提交任务的节点不能和spark工作节点交互,因为提交完任务后提交任务节点上会起一个进程,展示任务进度,大多端口为4044,工作节点需要反馈进度给该该端口,所以如果主机名或者IP在hosts中配置不正确,就会报

WARN YarnClientClusterScheduler: Initial job has not accepted any resources; check your cluster UI to ensure that workers are registered and have sufficient memory错误。
所以请检查主机名和IP是否配置正确。



我自己的理解为,程序提交任务到yarn后,会上传SPARK_JAR和SPARK_YARN_APP_JAR到hadoop节点, yarn根据任务情况来分配资源,在nodemanage节点上来启动org.apache.spark.deploy.yarn.WorkerLauncher工作节点来执行spark任务,执行完成后退出。

来自:http://my.oschina.net/132722/blog/220096

 本文由用户 jopen 自行上传分享,仅供网友学习交流。所有权归原作者,若您的权利被侵害,请联系管理员。
 转载本站原创文章,请注明出处,并保留原始链接、图片水印。
 本站是一个以用户分享为主的开源技术平台,欢迎各类分享!
 本文地址:https://www.open-open.com/lib/view/open1397180071590.html
分布式/云计算/大数据 Spark