Spark-3:Spark on yarn

Posted 2021-05-01 健哥说编程

  声明：【原创作品，版权所有。】

Spark on yarn

Spark已经可以配置运行在yarn上，只要在spark-env.sh中配置：

HADOOP_CONF_DIR为hadoop配置文件的目录即可。在配置完成以后，通过spark-submit提交任务：

spart-submit  \

 --master yarn \

--deploy-mode clustor \

YourApp.jar args

注意，如果配置spark on yarn，则spark会与NodeManager共同使用节点上的内存资源。有可能会有内存不足的问题，导致退出，解决的方式两个：

1：配置Spark所使用的内内存的核心类，在spark-env.sh中添加以下配置：

export SPARK_WORKER_MAMORY=512M

export SPARK_WORKER_CORES=1

        2：配置hadoop的配置文件yarn-site.xml中配置不让namenode检查内存

     <property>

                <!--配置是否检查物理内存-->

                <name>yarn.nodemanager.pmem-check-enabled</name>

                <value>false</value>

        </property>

        <property>

                <!--配置是否检查虚拟内存是否足够-->

                <name>yarn.nodemanager.vmem-check-enabled</name>

                <value>false</value>

        </property>

以下开始配置spark运行在yarn上。

为了更好的控制资源的使用，如在ResourceManager上，最好不要启动Spark的后台进程。

所以，如果是在非高可靠环境下，应该有一台独立的主机，用于启动NameNode和Resourcemanager，集群设计如下：

ip/主机名	软件	设计进程	说明
192.168.56.100 hadoop100	hadoop spark	NameNode ResourceManager SecondaryNameNode   SparkSubmit	做为HDFS和Yarn的主服务器
192.168.56.101 hadoop101	hadoop Spark	DataNode NodeManager   CoarseGrainedExecutorBackend（Spark Yarn的后台进程）
192.168.56.102 hadoop102	hadoop spark	DataNode NodeManager   ExecutorLauncher(随机出现在某个NodeManager节点上) CoarseGrainedExecutorBackend
192.168.56.103 hadoop103	hadoop spark	DataNode NodeManager CoarseGrainedExecutorBackend

说明，对于CoarseGrainedExecutorBackend进程，默认将会启动2个，这个选项在spark-env.sh中配置的默认值，可以通过export SPARK_EXECUTOR_INSTANCE=N来设置。但，最好的方案是在执行spark-shell或是spart-submit时，通过参数：--num-executors 3来指定。

如：

spark-shell --master yarn --num-executors 3

或

spark-submit --master yarn --num-executors 3

这样将会开启3个CoarseGrainedExecutorBackend这样的进程，如果仅为三台从节点，并设置--num-executors 5也是可以的，将会在些主机节点上，开启多个CoarseGrainedExecutorBackend进程，就如：

$ jps

NodeManager

DataNode

CoarseGrainedExecutorBackend

CoarseGrainedExecutorBackend

 

步1：在spark-env.sh中配置

在SPARK_HOME/conf目录下，修改spark-env.sh文件，配置如下：

HADOOP_CONF_DIR=/hadoop/hadoop-2.7.3/etc/hadoop

SPARK_WORKER_CORES=2   每一个工作节点的进程的数量  

SPARK_EXECUTOR_MEMTORY=512M

SPARK_DEIVER_MEMORY=512M

步2：将文件scp到其他节点

$ scp -r spark /  hadoop101:/spark/

 

步3：启动Spark-Shell

$ spark-shell --master yarn --num-executors 3

启动以后，检查各节点的进程，就会出现上表中所示的进程。

 

步4：使用spark-submit来提交scala程序

现在可以将Scala开发的程序，提交到Spark的yarn集群：

代码：

package cn.wang
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}
/**
  * 提交到Spark Yarn的代码示例
  */
object Spark07YarnWordCount {
  def main(args: Array[String]): Unit = {
    //接收输出和输出的参数
    if (args.length < 2) {
      println("请输入读取的文件，和输出的目录");
      return;
    }
    //声明SparkConfig
    val config: SparkConf = new SparkConf();
    config.setAppName("Spark07YarnWordCount");
    //设置一些其他的参数
    config.set("spark.executor.memory", "512M");
    //声明SparkContext对象
    val sc: SparkContext = new SparkContext(config);
    //读取文件,hdfs文件,返回的是HadoopRDD见源代码 SparkContext第999行
    val rdd1: RDD[String] = sc.textFile(args.apply(0));
    //现在开始计算行数
    val count: Long = rdd1.count();
    println("文件中的行数是：" + count);
    //进行单词统计
    rdd1.flatMap(_.split("\\s+")).map((_, 1)).reduceByKey((a, b) => a + b). //
      map(kv => {
      val str: String = kv._1 + "\t" + kv._2;
      str;
    }).saveAsTextFile(args.apply(1));
    //直接保存到hdfs上,必须要保证这个目录不存在，可以使用hdfs操作删除或是创建
    println("保存到HDFS结束")
    Thread.sleep(1000 * 60 * 3); //休眠3份钟-可选的
    sc.stop();
  }
}

 

现在将这个程序打成jar包：

然后将这个jar包上传到 Linux并通过spart-submit提交，或是开发一个linux shell脚本：

#!/bin/bash

if [ $# -eq 0 ]; then

   echo "请输入两个参数"

else

        spark-submit \

        --num-executors 3 \

        --master yarn \

        --class cn.wang.Spark07YarnWordCount \

        SparkDemo01.jar $1 $2

fi

 

也可以直接执行spark-submit命令：

spark-submit \

        --num-executors 3 \

        --master yarn \

        --class cn.wang.Spark07YarnWordCount \

        SparkDemo01.jar  /wang/a.txt  /out001

 

等待它执行完成，查看hdfs上的数据，如下显示，则说明OK！

$ hdfs dfs -cat /out001/*

Hello   4

jerry   1

Mary    2

Jack    1

 

 

 

2.5、总结

1：Spark有三种运行方式分别是：

Local 即本地模式，这种模式下只需要将Spark解压并执行spark-shell即会启动一个application，并可以通过4040查看Spark的运行状态。

运行在yarn/mesos集群上。本文档前面的内容，讲的是运行在yarn上。配置yarn集群，主要在spark_home/conf目录下，修改spark-env.sh文件。并添加环境变量：export  HADOOP_CONF_DIR=...。注意在虚拟机环境下，应该控制以下内容

2：三种方式driver的方式有所不同

local :  spark-shell --master local[1]

standalone : spark-shell --master spark://ip:7077

yarn : spark-shell --master yarn --deploy-mode cluster

3：使用spark-shell或是使用spark-submit都是启动一个driver也是application。

4：可以通过参数，或是配置的方式，指定excutors的个数及每一个executor的cores数量。

--num-executors 用于指定有多少executors

export SPARK_WORKER_CORES=N 可以指定每一个executor的cores数量。

   下一章：RDD算子。