Spark - 安装部署详细教程5（ON YARN模式）

五、ON YARN 模式

1，模式介绍

（2）假设我们的 Hadoop 集群是 node1、node2、node3。那我们可以选择把 Spark 部署在一个单独的节点上就可以了。而关于 Hadoop 集群的部署，可以参考我之前写的文章：

2，Spark 部署

wget https://archive.apache.org/dist/spark/spark-3.4.1/spark-3.4.1-bin-hadoop3.tgz

（2）然后执行下面命令将下载下来的压缩包解压至指定目录（目录地址可以根据需要自行调整）：

tar -zxvf spark-3.4.1-bin-hadoop3.tgz -C /usr/local

（3）进入 spark 目录中的 conf 目录，创建一个 spark-env.sh 文件：

cd /usr/local/spark-3.4.1-bin-hadoop3/conf/
vi spark-env.sh

（4）文件内容如下，指定 Hadoop 的配置文件目录：

export HADOOP_CONF_DIR=/usr/local/hadoop/etc/hadoop

3，提交任务

cd /usr/local/spark-3.4.1-bin-hadoop3

（2）然后执行如下命令通过这个 spark 客户点节点，向 Hadoop 集群上提交 spark 任务：

./bin/spark-submit \
--master yarn \
--class org.apache.spark.examples.SparkPi \
examples/jars/spark-examples_2.12-3.4.1.jar \
10

（3）执行完毕后控制台会输出 Pi 的值：

附：开启 Spark history Server（Spark 历史服务）

2，操作步骤

cd /usr/local/spark-3.4.1-bin-hadoop3/conf
vi spark-defaults.conf

• 文件的内容如下：

spark.eventLog.enabled true
spark.eventLog.dir hdfs://node1:9000/spark-log
spark.history.fs.logDirectory=hdfs://node1:9000/spark-log
spark.yarn.historyServer.address=http://node1:18080

（2） 接着编辑 conf 目录下的 spark-env.sh 文件：

vi spark-env.sh

• 在文件中添加如下日志配置：

export SPARK_HISTORY_OPTS="
-Dspark.history.ui.port=18080
-Dspark.history.fs.logDirectory=hdfs://node1:9000/spark-log"

（3）最后执行如下命令启动历史服务：

./sbin/start-history-server.sh

（4）然后我们可以使用 jps 命令验证下服务是否启动成功：

3，使用测试

cd /usr/local/hadoop/
bin/mapred --daemon start historyserver

（2）接着我们编写个简单的 WordCount 程序并达打成 Jar 包上传到服务器，其中我们会通过 foreach 将结果数据打印出来：

import org.apache.spark.{SparkConf, SparkContext}

object WordCount {
  def main(args: Array[String]): Unit = {
    //第一步：创建SparkContext
    val conf = new SparkConf()
    conf.setAppName("WordCount")//设置任务名称
    //  .setMaster("local")//local表示在本地执行
    val sc = new SparkContext(conf)

    //第二步：加载数据
    var path = "D:\\hello.txt"
    if (args.length == 1) {
      path = args(0)
    }
    val linesRDD = sc.textFile(path)

    //第三步：对数据进行切割，把一行数据切分成一个一个的单词
    val wordsRDD = linesRDD.flatMap(_.split(" "))

    //第四步：迭代words,将每个word转化为(word,1)这种形式
    val pairRDD = wordsRDD.map((_,1))

    //第五步：根据key(其实就是word)进行分组聚合统计
    val wordCountRDD = pairRDD.reduceByKey(_ + _)

    //第六步：将结果打印到控制台
    wordCountRDD.foreach(wordCount=>println(wordCount._1+"--"+wordCount._2))

    //第七步：停止SparkContext
    sc.stop()
  }
}

（3）然后执行如下命令使用 on yarn 模式向集群提交任务：

./bin/spark-submit \
--master yarn \
--class WordCount  \
spark-1.0-SNAPSHOT-jar-with-dependencies.jar \
hdfs://node1:9000/hello.txt

(4）执行完毕后，我们在 YARN 的 8088 界面也可以看到刚提交上去的任务信息，点击“History”链接。