最新Spark入门篇

一、Spark简介

1.什么是Spark

Apache Spark是一种快速的集群计算技术，基于Hadoop MapReduce技术，扩展了MapReduce模型，主要特性是在内存中集群计算，速度更快。即使在磁盘上进行复杂计算，Spark依然比MapReduce更加高效。另一方面，Apache Spark扩展了MapReduce模型以使用更多类型的计算。

1.1 使用基于Hadoop的Spark

Spark与Hadoop是兼容的，Hadoop组件可以通过以下方式与Spark一起使用：

HDFS

Spark可以在HDFS之上运行，以利用分布式存储
MapReduce

Spark可以与MapReduce一起用于同一个Hadoop集群，也可以单独作为处理框架使用
YARN

可以使Spark应用程序在YARN(Hadoop NextGen)上运行
批处理和实时处理

MapReduce和Spark一起使用，其中MapReduce用于批处理，Spark用于实时处理

1.2 Spark的组件

spark的组件主要有以下几种：

Spark Core

Spark Core是大规模并行和分布式数据处理的基础引擎。核心是分布式执行引擎，Java，Scala和Python API为分布式ETL应用程序开发提供了一个平台。此外，在核心上构建的其他库允许用于流式传输，SQL和机器学习的各种工作负载。它负责：

1.内存管理和故障恢复

2.在群集上调度，分发和监视作业

3.与存储系统交互
Spark Streaming

Spark Streaming是Spark的组件，用于处理实时流数据。因此，它是核心Spark API的补充。它支持实时数据流的高吞吐量和容错流处理。基本流单元是DStream，它基本上是一系列处理实时数据的RDD(弹性分布式数据集)
Spark SQL

Spark SQL是Spark中的一个新模块，它使用Spark编程API实现集成关系处理。它支持通过SQL或Hive查看数据。对于那些熟悉RDBMS的人来说，Spark SQL将很容易从之前的工具过度到可以扩展传统关系数据处理的边界。

Spark SQL通过函数编程API集成关系处理。此外，它为各种数据源提供支持，并且使用代理转换编织SQL查询，从而产生一个非常强大的工具。

Spark SQL包含四个库：

1.Data Source API

2.DataFrame API

3.Interpreter & Optimizer

4.SQL Service
GraphX

GraphX是用于图形和图形并行计算的Spark API。因此，它使用弹性分布式属性图扩展了Spark RDD。

属性图是一个有向多图，它可以有多个平行边。每个边和定点都有与之关联的用户定义属性。这里，平行边缘允许相同顶点之间的多个关系。在高层次上，GraphX通过引入弹性分布式属性图来扩展Spark RDD抽象：一个定向多图，其属性附加到每个顶点和边。

为了支持图形计算，GraphX公开了一组基本运算符(如，subgraph，joinVertices和mapReduceTriplets)以及Pregel API的优化变体。此外，GraphX包含越来越多的图算法和构建起，以简化图形分析任务。
MLlib(Machine Learning)

MLib代表机器学习库，用于在Apache Spark中执行机器学习功能。

2.Spark运行模式

5种模式：

local

本地模式，主要用于本地开发测试
Standlone

集群模式，典型的Master/slave模式
on yarn

集群模式，运行在yarn资源管理器框架之上，由yarn负责资源管理，Spark负责任务调度和计算
on mesos

集群模式，运行在mesos资源管理器框架之上，由mesos负责资源管理，Spark负责任务调度和计算
Kubernetes (experimental)

Kubernetes 提供以容器为中心的基础设施的开源平台

3.Spark运行流程

如上图所示，spark的运行流程为：

Spark通过SparkContext向Cluster manager(资源管理器)申请所需执行的资源（cpu、内存等）
Cluster manager分配应用程序执行需要的资源，在Worker节点上创建Executor
SparkContext 将程序代码（jar包或者python文件）和Task任务发送给Executor执行，并收集结果给Driver。

二、Spark安装

主机环境：ubuntu

1.下载文件

进入https://www.apache.org下载spark-2.4.3-bin-hadoop2.7.tgz

下载文件后，放到ubuntu的software文件夹

# 切换到software目录
zifan@ubuntu:~$ cd software/
zifan@ubuntu:~/software$ tar xvf spark-2.4.3-bin-hadoop2.7.tgz

移动spark文件夹到/usr/local/spark目录

# 移动文件夹使用sudo mv指令
zifan@ubuntu:~/software$ sudo mv spark-2.4.3-bin-hadoop2.7 /usr/local/spark

切换到/usr/local/spark目录查看

# 切换到/usr/local/spark目录
zifan@ubuntu:~/software$ cd /usr/local/spark
zifan@ubuntu:/usr/local/spark$ ls
bin   data      jars        LICENSE   NOTICE  R          RELEASE  yarn
conf  examples  kubernetes  licenses  python  README.md  sbin

2.配置Spark环境变量

zifan@ubuntu:/usr/local/spark$ sudo vi ~/.bashrc
# 打开文件，最后一行加上
export PATH=$PATH:/usr/local/spark/bin

使配置生效

zifan@ubuntu:/usr/local/spark$ source ~/.bashrc

3.验证Spark安装

在控制台输入spark-shell指令

zifan@ubuntu:/usr/local/spark$ spark-shell

如果出现以下输出结果，则表示spark安装成功。

Spark session available as 'spark'.
Welcome to
      ____              __
     / __/__  ___ _____/ /__
    _ / _ / _ `/ __/  '_/
   /___/ .__/\_,_/_/ /_/\_   version 2.4.3
      /_/

Using Scala version 2.11.12 (Java HotSpot(TM) 64-Bit Server VM, Java 1.8.0_171)
Type in expressions to have them evaluated.
Type :help for more information.

scala>

4.退出

使用Ctrl+D或 :quit退出

三、使用Java开发本地Spark应用

1. 操作步骤

以下表述为在ubuntu环境里面操作的记录。

安装maven

https://maven.apache.org，选择合适的版本下载，这里选择的是apache-maven-3.6.1-bin.tar.gz版本

下载文件后，放到ubuntu的software文件夹

# 切换到software目录
zifan@ubuntu:~$ cd software/
zifan@ubuntu:~/software$ tar zxvf apache-maven-3.6.1-bin.tar.gz

移动文件夹到/usr/local/maven目录

# 移动文件夹使用sudo mv指令
zifan@ubuntu:~/software$ sudo mv apache-maven-3.6.1 /usr/local/maven

切换到/usr/local/maven目录查看

# 切换到/usr/local/sbt
zifan@ubuntu:~/software$ cd /usr/local/maven
zifan@ubuntu:/usr/local/maven$ ls
bin  boot  conf  lib  LICENSE  NOTICE  README.txt

配置环境变量

zifan@ubuntu:~/sparkapp2$ sudo vim ~/.bashrc

在文件末尾添加

export M2_HOME=/usr/local/maven
export CLASSPATH=$CLASSPATH:$M2_HOME/lib
export PATH=$PATH:$M2_HOME/bin

保存后退出

使环境变量生效

zifan@ubuntu:~/sparkapp2$ source ~/.bashrc

测试

zifan@ubuntu:~/sparkapp2$ mvn -v

显示内容表示安装成功

Apache Maven 3.6.1 (d66c9c0b3152b2e69ee9bac180bb8fcc8e6af555; 2019-09-05T03:00:29+08:00)
Maven home: /usr/local/maven
Java version: 1.8.0_171, vendor: Oracle Corporation, runtime: /home/zifan/software/jdk1.8.0_171/jre
Default locale: en_US, platform encoding: UTF-8
OS name: "linux", version: "4.4.0-142-generic", arch: "amd64", family: "unix"

编写Java应用程序

1.创建文件夹sparkapp2作为应用程序根目录

# 逐级创建目录
zifan@ubuntu:~$ mkdir -p ./sparkapp2/src/main/java

2.在./sparkapp2/src/main/java目录下创建SparkTest1.java

zifan@ubuntu:~$ vim ./sparkapp2/src/main/java/SparkTest1.java

写入以下内容

/*** SparkTest1.java ***/
    import org.apache.spark.api.java.*;
    import org.apache.spark.SparkConf;
    import org.apache.spark.api.java.function.Function;
 
    public class SparkTest1 {
        public static void main(String[] args) {
            String logFile = "file:///usr/local/spark/README.md"; // 此文件为安装时的说明文件
            SparkConf conf=new SparkConf().setAppName("SparkTest1");
            JavaSparkContext sc=new JavaSparkContext(conf);
            JavaRDD<String> logData = sc.textFile(logFile).cache();
 
            long A1 = logData.filter(new Function<String, Boolean>() {
                public Boolean call(String s) { return s.contains("spark"); }
            }).count();
 
            long B1 = logData.filter(new Function<String, Boolean>() {
                public Boolean call(String s) { return s.contains("test"); }
            }).count();
 
            System.out.println("Lines with spark: " + A1 + ", lines with test: " + B1);
        }
    }

该程序依赖Spark Java API，需要通过Maven进行编译打包。

3.在./sparkapp2中新建文件pom.xml

zifan@ubuntu:~$ vim ./sparkapp2/pom.xml

写入内容

        <project>
        <groupId>com.zifan.example</groupId>
        <artifactId>sparktest1-project</artifactId>
        <modelVersion>4.0.0</modelVersion>
        <name>Spark Project</name>
        <packaging>jar</packaging>
        <version>1.0</version>
        <repositories>
        <repository>
            <id>maven-ali</id>
            <url>http://maven.aliyun.com/nexus/content/groups/public//</url>
            <releases>
                <enabled>true</enabled>
            </releases>
            <snapshots>
                <enabled>true</enabled>
                <updatePolicy>always</updatePolicy>
                <checksumPolicy>fail</checksumPolicy>
            </snapshots>
        </repository>
    </repositories>
        <dependencies>
            <dependency> <!-- Spark依赖，注意version为spark版本 -->
                <groupId>org.apache.spark</groupId>
                <artifactId>spark-core_2.11</artifactId>
                <version>2.4.3</version>
            </dependency>
        </dependencies>
    </project>

使用maven打包java程序

1.为了保证maven能够正常运行，先执行如下命令检查整个应用程序的文件结构
```
zifan@ubuntu:~$ cd sparkapp2
zifan@ubuntu:~/sparkapp2$ find .
# 结果显示
.
./pom.xml
./src
./src/main
./src/main/java
./src/main/java/SparkTest1.java
```
2.使用mvn打包
```
# sparkapp2目录下执行
zifan@ubuntu:~/sparkapp2$ mvn package
```
结果显示
```
Downloaded from central: https://repo.maven.apache.org/maven2/org/codehaus/plexus/plexus-io/2.0.2/plexus-io-2.0.2.jar (58 kB at 3.6 kB/s)
Downloaded from central: https://repo.maven.apache.org/maven2/org/codehaus/plexus/plexus-archiver/2.1/plexus-archiver-2.1.jar (184 kB at 4.0 kB/s)
[INFO] Building jar: /home/zifan/sparkapp2/target/sparktest1-project-1.0.jar
[INFO] ------------------------------------------------------------------------
[INFO] BUILD SUCCESS
[INFO] ------------------------------------------------------------------------
[INFO] Total time:  0:45 min
[INFO] Finished at: 2019-09-20T20:26:15+08:00
[INFO] ------------------------------------------------------------------------
```
至此jar包完成，下面讲述通过spark运行的方式，spark运行jar包通过spark-submit命令进行。

spark-submit的命令格式：
```
./bin/spark-submit 
  --class <main-class>  //需要运行的程序的主类，应用程序的入口点
  --master <master-url>  //Master URL，下面会有具体解释
  --deploy-mode <deploy-mode>   //部署模式
  ... # other options  //其他参数
  <application-jar>  //应用程序JAR包
  [application-arguments] //传递给主类的主方法的参数
```
deploy-mode这个参数用来指定应用程序的部署模式，部署模式有两种：client和cluster，默认是client。

这两种模式的区别就是是否在本地运行程序，client模式会在本地运行；而cluster模式时，则不会，该模式一般会在Worker节点上运行程序。

Spark的运行模式取决于传递给SparkContext的Master URL的值。Master URL可以选择以下其中一种形式：

1.local 使用一个Worker线程本地化运行SPARK(完全不并行)

2.local[*] 使用逻辑CPU个数数量的线程来本地化运行Spark

3.local[K] 使用K个Worker线程本地化运行Spark

4.spark://HOST:PORT 连接到指定的Spark standalone master，本地独立集群运行，默认端口是7077

5.yarn-client 以客户端模式连接YARN集群。集群的位置可以在HADOOP_CONF_DIR 环境变量中找到

6.yarn-cluster 以集群模式连接YARN集群。集群的位置可以在HADOOP_CONF_DIR 环境变量中找到

7.mesos://HOST:PORT 连接到指定的Mesos集群。默认接口是5050

从上可以看出，本地化运行模式主要有前四种。前三种都是local模式，第四种为standalone模式。因此本地模式大致可以分为local和standalone 两个大类。

2. 本地local模式

运行程序

将生成的jar包通过spark-submit提交到Spark中运行，命令如下

# --master local[4]表示采用本地模式，在4个CPU核心上运行spark-shell
zifan@ubuntu:~/sparkapp$ /usr/local/spark/bin/spark-submit --class "SparkTest1" --master local[4] ~/sparkapp2/target/sparktest1-project-1.0.jar

最终显示结果

Lines with spark: 13, lines with test: 8

这个程序是计算一个文本文件中包含”spark"的行数和包含“test”的行数，注意统计是区分大小写的。

运行结果截图如下

3.本地独立集群模式

本地独立集群方式（Spark standalone），这里主要讲下本地独立集群模式的web-ui页面

启动Master

zifan@ubuntu:~$ cd /usr/local/spark
zifan@ubuntu:/usr/local/spark$ sbin/start-master.sh

打开http://ubuntu主机IP地址:8080/,界面如下:

启动Slave

注意图中的url

zifan@ubuntu:/usr/local/spark$ sbin/start-slave.sh spark://ubuntu:7077

启动成功后，刷新下页面，可以看到Alive Workers已经有值了

将jar包放置于spark中运行

# --master spark://ubuntu:7077表示采用本地独立集群模式运行spark-shell
zifan@ubuntu:~/sparkapp$ /usr/local/spark/bin/spark-submit --class "SparkTest1" --master spark://ubuntu:7077 ~/sparkapp2/target/sparktest1-project-1.0.jar

可以看到在Completed Applications栏多了一条记录，这条记录即为刚执行完的程序。

最终显示结果

Lines with spark: 13, lines with test: 8

运行结果截图如下

4.总结

上面的例子应该来说是比较简单的，spark框架优点在于处理大数据时的高效率，特别是对于百万级以上的数据，spark有着惊人的处理效率。