Spark - OnYARN 模式搭建，并使用 Scala、Java、Python 三种语言测试

一、SparkOnYarn搭建

安装前需要提前安装好 hadoop 环境，关于 HDFS 和 Yarn 集群的搭建可以参考下面我的博客：

https://blog.csdn.net/qq_43692950/article/details/127158935

下面是我 Hadoop 的安装结构

开始前请确保 hadoop 已经成功启动起来。

准备一台服务器，可以和 Hadoop 装在一台机子上，下面有需要读取 Hadoop 的配置，如果是台纯净的服务，需要将 Hadoop 的配置文件放上去。Yarn 需要配置历史服务器，如果没有配置通过修改下面文件开启：

修改 Hadoop 安装目录 etc/hadoop/yarn-site.xml 文件：

vim yarn-site.xml
1

添加下面配置：

<property>
    <name>yarn.log.server.urlname>
    <value>http://node1:19888/jobhistory/logsvalue>
property>

<property>
    <name>yarn.nodemanager.pmem-check-enabledname>
    <value>falsevalue>
property>
<property>
    <name>yarn.nodemanager.vmem-check-enabledname>
    <value>falsevalue>
property>
1
2
3
4
5
6
7
8
9
10
11
12
13
14

该需要同步到集群的其他节点：

scp /export/server/hadoop-3.1.4/etc/hadoop/yarn-site.xml root@node2:/export/server/hadoop-3.1.4/etc/hadoop/yarn-site.xml
scp /export/server/hadoop-3.1.4/etc/hadoop/yarn-site.xml root@node3:/export/server/hadoop-3.1.4/etc/hadoop/yarn-site.xml
1
2

重启 Hadoop 集群：

stop-all.sh

start-all.sh
1
2
3

下载 Spark 安装包，这里我用的 3.0.1 版本：

http://spark.apache.org/downloads.html

先将下载后的安装包上传至 node1 节点，解压安装包：

tar -zxvf spark-3.0.1-bin-hadoop2.7.tgz
1

进入到加压目录下的 conf 下，修改配置：

修改 spark-defaults.conf：

mv spark-defaults.conf.template spark-defaults.conf
vi spark-defaults.conf
1
2

添加内容：

spark.eventLog.enabled                  true
spark.eventLog.dir                      hdfs://node1:8020/sparklog/
spark.eventLog.compress                 true
spark.yarn.historyServer.address        node1:18080
spark.yarn.jars                         hdfs://node1:8020/spark/jars/*
1
2
3
4
5

修改 spark-env.sh ：

vi spark-env.sh
1

## 设置JAVA安装目录
JAVA_HOME=/usr/lib/jvm/java-1.8.0

## HADOOP软件配置文件目录，读取HDFS上文件和运行Spark在YARN集群时需要
HADOOP_CONF_DIR=/export/server/hadoop-3.1.4/etc/hadoop
YARN_CONF_DIR=/export/server/hadoop-3.1.4/etc/hadoop

## 配置spark历史日志存储地址
SPARK_HISTORY_OPTS="-Dspark.history.fs.logDirectory=hdfs://node1:8020/sparklog/ -Dspark.history.fs.cleaner.enabled=true"
1
2
3
4
5
6
7
8
9

这里的 sparklog 需要手动创建：

hadoop fs -mkdir -p /sparklog
1

修改日志的级别：

mv log4j.properties.template log4j.properties
vi log4j.properties
1
2

配置Spark 的依赖 jar 包，将 jar 都上传到 HDFS 中：

在HDFS上创建存储spark相关jar包的目录：

hadoop fs -mkdir -p /spark/jars/
1

上传$SPARK_HOME/jars所有jar包到HDFS：

hadoop fs -put /export/spark/spark-3.0.1-bin-hadoop2.7/jars/* /spark/jars/
1

启动MRHistoryServer服务，在node1执行命令

mr-jobhistory-daemon.sh start historyserver
1

进入到 Spark 解压目录，启动Spark HistoryServer服务：

sbin/start-history-server.sh
1

MRHistoryServer服务WEB UI页面：

http://node1:19888

Spark HistoryServer服务WEB UI页面：

http://node1:18080/

二、使用三种语言测试环境

1. java 和 Scala 项目

创建一个普通的Maven项目，在 pom 中添加 Scala 和 Spark 的依赖：

<dependency>
    <groupId>org.scala-langgroupId>
    <artifactId>scala-libraryartifactId>
    <version>2.12.11version>
dependency>


<dependency>
    <groupId>org.apache.sparkgroupId>
    <artifactId>spark-core_2.12artifactId>
    <version>3.0.1version>
dependency>
1
2
3
4
5
6
7
8
9
10
11
12
13

在 main 下面创建 scala 包，专门存放 scala 程序，java 下专门存放 java 程序：

Scala 测试程序

object RddTestScala {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setAppName("spark")
    val sc = new SparkContext(conf)
    sc.setLogLevel("WARN")

    val texts= sc.parallelize(Seq("abc", "abc", "ff", "ee", "ff"))
    val counts = texts.map((_,1)).reduceByKey(_ + _)
    println(counts.collectAsMap())
  }
}
1
2
3
4
5
6
7
8
9
10
11

Java 测试程序

public class RddTestJava {
    public static void main(String[] args) {
        SparkConf conf = new SparkConf().setAppName("spark");
        JavaSparkContext sc = new JavaSparkContext(conf);
        sc.setLogLevel("WARN");

        JavaRDD<String> texts= sc.parallelize(Arrays.asList("abc", "abc", "ff", "ee", "ff"));
        JavaPairRDD<String, Integer> counts = texts.mapToPair(s -> new Tuple2<>(s, 1)).reduceByKey(Integer::sum);
        System.out.println(counts.collectAsMap());
    }
}
1
2
3
4
5
6
7
8
9
10
11

由于默认情况下使用 maven 编译不会编译 scala 程序，在 build 中添加 scala 的插件：

<build>
    <sourceDirectory>src/main/javasourceDirectory>
    <plugins>
        
        <plugin>
            <groupId>org.apache.maven.pluginsgroupId>
            <artifactId>maven-compiler-pluginartifactId>
            <version>3.5.1version>
            <configuration>
                <source>8source>
                <target>8target>
            configuration>
        plugin>
        
        <plugin>
            <groupId>net.alchim31.mavengroupId>
            <artifactId>scala-maven-pluginartifactId>
            <version>3.2.2version>
            <executions>
                <execution>
                    <goals>
                        <goal>compilegoal>
                        <goal>testCompilegoal>
                    goals>
                    <configuration>
                        <args>
                            <arg>-dependencyfilearg>
                            <arg>${project.build.directory}/.scala_dependenciesarg>
                        args>
                    configuration>
                execution>
            executions>
        plugin>
        <plugin>
            <groupId>org.apache.maven.pluginsgroupId>
            <artifactId>maven-surefire-pluginartifactId>
            <version>2.18.1version>
            <configuration>
                <useFile>falseuseFile>
                <disableXmlReport>truedisableXmlReport>
                <includes>
                    <include>**/*Test.*include>
                    <include>**/*Suite.*include>
                includes>
            configuration>
        plugin>
        <plugin>
            <groupId>org.apache.maven.pluginsgroupId>
            <artifactId>maven-shade-pluginartifactId>
            <version>2.3version>
            <executions>
                <execution>
                    <phase>packagephase>
                    <goals>
                        <goal>shadegoal>
                    goals>
                    <configuration>
                        <filters>
                            <filter>
                                <artifact>*:*artifact>
                                <excludes>
                                    <exclude>META-INF/*.SFexclude>
                                    <exclude>META-INF/*.DSAexclude>
                                    <exclude>META-INF/*.RSAexclude>
                                excludes>
                            filter>
                        filters>
                        <transformers>
                            <transformer
                                    implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer">
                                <mainClass>com.bxc.RddTestJavamainClass>
                            transformer>
                        transformers>
                    configuration>
                execution>
            executions>
        plugin>
    plugins>
build>
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79

打成 jar 包：

mvn clean package
1

将打包后的jar包上传至 node1 节点，进到 Spark 解压目录下：

运行 Scala 脚本

bin/spark-submit \
--master yarn \
--deploy-mode client \
--driver-memory 512m \
--driver-cores 1 \
--executor-memory 512m \
--num-executors 1 \
--executor-cores 1 \
--class com.bxc.RddTestScala \
/export/spark/spark-submit-demo-1.0-SNAPSHOT.jar
1
2
3
4
5
6
7
8
9
10

运行 Java 脚本

bin/spark-submit \
--master yarn \
--deploy-mode client \
--driver-memory 512m \
--driver-cores 1 \
--executor-memory 512m \
--num-executors 1 \
--executor-cores 1 \
--class com.bxc.RddTestJava \
/export/spark/spark-submit-demo-1.0-SNAPSHOT.jar
1
2
3
4
5
6
7
8
9
10

2. Python项目

编写 Python 脚本：

from pyspark import SparkConf, SparkContext

if __name__ == '__main__':
    conf = SparkConf().setAppName('spark')
    sc = SparkContext(conf=conf)
    sc.setLogLevel("WARN")

    texts = sc.parallelize(["abc", "abc", "ff", "ee", "ff"])
    counts = texts.map(lambda s:(s, 1)).reduceByKey(lambda v1,v2:v1+v2)
    print(counts.collectAsMap())
1
2
3
4
5
6
7
8
9
10

将脚本上传至 node1 节点，，进到 Spark 解压目录下：

bin/spark-submit \
--master yarn \
--deploy-mode client \
--driver-memory 512m \
--driver-cores 1 \
--executor-memory 512m \
--num-executors 1 \
--executor-cores 1 \
/export/spark/RddTestPy.py
1
2
3
4
5
6
7
8
9

查看web界面：

http://node1:8088/cluster

三、YARN 模式下的两种运行模式

1. client

例如：

bin/spark-submit \
--master client\
--deploy-mode cluster \
--driver-memory 512m \
--driver-cores 1 \
--executor-memory 512m \
--num-executors 1 \
--executor-cores 1 \
--class com.bxc.RddTestJava \
/export/spark/spark-submit-demo-1.0-SNAPSHOT.jar
1
2
3
4
5
6
7
8
9
10

2. cluster

例如：

bin/spark-submit \
--master cluster\
--deploy-mode cluster \
--driver-memory 512m \
--driver-cores 1 \
--executor-memory 512m \
--num-executors 1 \
--executor-cores 1 \
--class com.bxc.RddTestJava \
/export/spark/spark-submit-demo-1.0-SNAPSHOT.jar
1
2
3
4
5
6
7
8
9
10