apachespark官网（如何使用spark将程序提交任务到yarn-Spark-about云开发）

本文目录

• 如何使用spark将程序提交任务到yarn-Spark-about云开发
• org.apache.spark.rdd.rdd$怎么解决
• apache spark是什么
• org.apache.spark.api.java.optional在哪个包下
• Apache Spark 黑名单(Blacklist)机制介绍
• apache spark是什么意思

如何使用spark将程序提交任务到yarn-Spark-about云开发

使用脚本提交
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:// 表明是本地文件，如果使用hdfs上的文件将file://替换为hdfs://主机名:端口号。建议使用hdfs来引用 spark-assembly-0.9.0-incubating-hadoop2.2.0.jar，因为这个文件比较大，如果使用file://每次提交任务都需要上传这个jar到各个集群，很慢。
其中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任务，执行完成后退出。

org.apache.spark.rdd.rdd$怎么解决

如何创建RDD？RDD可以从普通数组创建出来，也可以从文件系统或者HDFS中的文件创建出来。举例：从普通数组创建RDD，里面包含了1到9这9个数字，它们分别在3个分区中。scala》 val a = sc.parallelize(1 to 9, 3)a: org.apache.spark.rdd.RDD = Array((1,2), (1,3), (1,4), (1,5), (3,4), (3,5))上述例子中原RDD中每个元素的值被转换为一个序列（从其当前值到5），比如第一个KV对(1,2), 其值2被转换为2，3，4，5。然后其再与原KV对中Key组成一系列新的KV对(1,2),(1,3),(1,4),(1,5)。reducereduce将RDD中元素两两传递给输入函数，同时产生一个新的值，新产生的值与RDD中下一个元素再被传递给输入函数直到最后只有一个值为止。举例scala》 val c = sc.parallelize(1 to 10)scala》 c.reduce((x, y) =》 x + y)res4: Int = 55上述例子对RDD中的元素求和。reduceByKey顾名思义，reduceByKey就是对元素为KV对的RDD中Key相同的元素的Value进行reduce，因此，Key相同的多个元素的值被reduce为一个值，然后与原RDD中的Key组成一个新的KV对。

apache spark是什么

Apache Spark是一个强大的开源处理引擎，最初由Matei Zaharia开发，是他在加州大学伯克利分校的博士论文的一部分。

Apache Spark是快速、易于使用的框架，允许你解决各种复杂的数据问题，无论是半结构化、结构化、流式，或机器学习、数据科学。网页链接

org.apache.spark.api.java.optional在哪个包下

***隐藏网址***
如果你只想导入包的话（了解不多，1.6版本可以导入google的optional，百度“com.google.common”，2.1版本已有自己的optional）：在spark-core_对应版本号.jar。

Apache Spark 黑名单(Blacklist)机制介绍

在使用 Apache Spark 的时候，作业会以分布式的方式在不同的节点上运行；特别是当集群的规模很大时，集群的节点出现各种问题是很常见的，比如某个磁盘出现问题等。我们都知道 Apache Spark 是一个高性能、容错的分布式计算框架，一旦它知道某个计算所在的机器出现问题（比如磁盘故障），它会依据之前生成的 lineage 重新调度这个 Task。

我们现在来考虑下下面的场景：

上面提到的场景其实对我们人来说可以通过某些措施来避免。但是对于 Apache Spark 2.2.0 版本之前是无法避免的，不过高兴的是，来自 Cloudera 的工程师解决了这个问题：引入了黑名单机制 Blacklist（详情可以参见 SPARK-8425 ，具体的设计文档参见 Design Doc for Blacklist Mechanism ），并且随着 Apache Spark 2.2.0 版本发布，不过目前还处于实验性阶段。

黑名单机制其实是通过维护之前出现问题的执行器（Executors）和节点（Hosts）的记录。当某个任务（Task）出现失败，那么黑名单机制将会追踪这个任务关联的执行器以及主机，并记下这些信息；当在这个节点调度任务出现失败的次数超过一定的数目（默认为2），那么调度器将不会再将任务分发到那台节点。调度器甚至可以杀死那台机器对应的执行器，这些都可以通过相应的配置实现。

我们可以通过 Apache Spark WEB UI 界面看到执行器的状态（Status）：如果执行器处于黑名单状态，你可以在页面上看到其状态为 Blacklisted ，否则为 Active。如下图所示：

拥有了黑名单机制之后，上面场景的问题就可以很好的解决。

目前黑名单机制可以通过一系列的参数来控制，主要如下：

因为黑名单机制目前还处于实验性状态，所以上面的一些参数可能会在后面的 Spark 中有所修改。

apache spark是什么意思

n.火花;火星;电火花;(指品质或感情)一星，丝毫，一丁点。
averysmallburningpieceofmaterialthatisproducedbysththatisburningorbyhittingtwohardsubstancestogether。
Asparkisatinybrightpieceofburningmaterialthatfliesupfromsomethingthatisburning.
***隐藏网址***