1. Spark 概述

1.1 Spark 介绍

1.1.1 Spark 是什么

  Spark 是一种基于内存的快速、通用、可扩展的大数据分析引擎。

1.1.2 Spark 的由来

1. Hadoop 的历史

   2003,2004 Google 2 篇论文

   （1）2011 年发布 1.0 版本
   （2）2012 年发布稳定版
   
   ·MR 的缺点：

     mr 基于数据集的计算，所以面向数据。

    ① 基本运算规则：从存储介质中获取（采集）数据，然后进行计算，最后将结果存储到介质中，所以主要应用于一次性计算，不适合数据挖掘和机器学习这样迭代计算和图形挖掘计算。
    ② MR 基于文件存储介质的操作，所以性能非常的慢。
    ③ MR 和 hadoop 紧密耦合一起，无法动态替换。

   （3）2013 年 10月发布 2.X 版本（Yarn）

2. Spark 历史

   2009 年诞生于加州大学伯克利分校 AMPLab，项目采用 Scala 编写。
   2010 年开源;
   2013 年 6 月成为 Apache 孵化项目
   2014 年 2 月成为 Apache 顶级项目。

   Spark 基于 hadoop1.X 架构思想，采用自己的方式改善 Hadoop1.X 中的问题。
   Spark 计算基于内存（多任务） ，并且基于 Scala 语言开发，所以天生适合迭代计算。
   

1.2 Spark 内置模块

1. Spark Core

   实现了 Spark 的基本功能，包括任务调度、内存管理、错误恢复、与存储系统交互等模块。Spark Core 中还包括对弹性分布数据集（Resilent Distributed DataSet，简称 RDD）的 API 定义。

2. Spark SQL

   是 Spark 用来操作结构化数据的程序包。通过 Spark SQL，我们可以使用 SQL 或者 Apache Hive 版本的 SQL 方言（HQL）来查询数据。Spark SQL 支持多种数据源，比如 Hive 表、Parquet 以及 JSON 等。

3. Spark Streaming

   是 Spark 提供的对实时数据进行流式计算的组件。提供了用来操作数据流的 API，并且与 Spark Core 中的 RDD API 高度对应。

4. Spark Mlib

   提供常见的机器学习（ML）功能的程序库。包括分类、回归、聚类、协同过滤等，还提供了模型评估、数据导入等额外支持功能。

5. 集群管理器

   Spark 设计可以高效地在一个计算节点到数千个计算节点之间伸缩计算。为了实现这样的要求，同时获得最大灵活性，Spark 支持在各种集群管理器（Cluster Manager）上运行，包括 Hadoop YARN、Apache Mesos，以及 Spark 自带的一个简易调度器，叫作独立调度器。

1.3 Spark 特点

1. 速度快

   与 Hadoop 的 MapReduce 相比，Spark 基于内存的运算要快 100 倍以上，基于硬盘的运算也要快 10 倍以上。Spark 实现了高效的 DAG 执行引擎，可以通过基于内存来高效处理数据流。计算的中间结果是存在于内存中的。

2. 易用

   Spark 支持 Java、Python 和 Scala 的 API，还支持超过 80 种高级算法，使用户可以快速构建不同的应用。而且 Spark 支持交互式的 Python 和 Scala 的 Shell，可以方便地在这些 Shell 中使用 Spark 集群来验证解决问题的方法。

3. 通用

   Spark 提供了统一的解决方案。Spark 可以用于批处理、交互式查询（Spark SQL）、实时流处理（Spark Streaming）、机器学习（Spark Mlib）和图计算（GraphX）。这些不同类型的处理都可以在同一应用中无缝使用。减少了开发和维护的人力成本和部署平台的物力成本。

4. 兼容性

   Spark 可以非常方便地与其他的开源产品进行融合。比如，Spark 可以使用 Hadoop YARN 和 Apache Mesos 作为它的资源管理和调度器，并且可以处理所有 Hadoop 支持的数据，包括 HDFS、HBase 等。这对于已经部署 Hadoop 集群的用户特别重要，因为不需要做任何数据迁移就可以使用 Spark 的强大处理能力。

1.4 Spark 的重要角色

1.4.1 Driver（驱动器）

   Spark 的驱动器是执行开发程序中 main 方法的进程。它负责开发人员编写的用来创建 SparkContent、创建 RDD，以及进行 RDD 的转化操作和行动操作代码的执行。当启动 Spark shell 的时候，系统后台自启动了一个 Spark 驱动程序，就是在 Spark shell 中预加载了一个叫作 sc 的 SparkContent 对象。如果驱动器程序终止，那么 Spark 应用也结束了。

Driver 主要负责：

1. 把用户程序转为作业（JOB）。
2. 跟踪 Executor 的运行状况。
3. 为执行器节点调度任务。
4. UI 展示应用运行状态。

1.4.2 Executor（执行器）

   Spark Executor 是一个工作进程，负责在 Spark 作业中运行任务，任务之间相互独立。Spark 应用启动时，Executor 节点被同时启动，并且始终伴随着整个 Spark 应用的生命周期而存在。如果有 Executor 节点发生了故障或崩溃，Spark 应用也可以继续进行，因为会将出错节点上的任务调度到其他 Executor 节点上继续运行。

Executor 主要负责：

1. 负责运行组成 Spark 应用的任务，并将结果返回个驱动器进程。
2. 通过自身的块管理器（Block Manager）为用户程序中要求缓存的 RDD 提供内存式存储。RDD 是直接缓存在 Executor 进程内的，因此任务可以在运行时充分利用缓存数据加速运算。

1.5 Spark 官网

  官网地址

2. Spark 运行模式

2.1 Local 模式

2.1.1 概述

  Local 模式就是运行在一台计算机上的模式，通常就是用于在本机上练手和测试。

  它可以通过以下几种方式设置 Master。

1. local

   所有计算都运行在一个线程当中，没有任何并行计算，通常我们在本机执行一些测试代码，或者练手，就用这种模式;

2. local[K]

   指定使用几个线程来运行计算，比如 local[4 ]就是运行 4 个Worker线程。通常我们的 Cpu 有几个 Core，就指定几个线程，最大化利用 Cpu 的计算能力;

3. local[*]

   这种模式直接帮你按照 Cpu 最多 Cores 来设置线程数了。

2.1.2 安装使用

1. 把 spark 的安装包上传到服务器上（spark-2.3.1-bin-hadoop2.7.tgz）
2. 解压 spark 安装包到指定目录

tar -zxvf spark-2.3.1-bin-hadoop2.7.tgz -C /hadoop/

3. 重命名解压后的目录

mv spark-2.3.1-bin-hadoop2.7/ spark-2.3.1

4. 配置环境变量

vim /etc/profile

    在文件中添加以下内容：

#SPARK
export SPARK_HOME=/hadoop/spark-2.3.1
export PATH=$PATH:$SPARK_HOME/bin

    使配置文件生效：

source /etc/profile

2.1.3 官方求 PI 案例

1. 输入命令

spark-submit \
--class org.apache.spark.examples.SparkPi \
--executor-memory 1G \
--total-executor-cores 2 \
/hadoop/spark-2.3.1/examples/jars/spark-examples_2.11-2.3.1.jar \
100

    ① 基本语法

spark-submit \
--class <main-class>
--master <master-url> \
--deploy-mode <deploy-mode> \
--conf <key>=<value> \
... # other options
<application-jar> \
[application-arguments]

    ② 参数说明：

参数	说明
class	你的应用的启动类 (如 org.apache.spark.examples.SparkPi)
master	指定 Master 的地址，默认为 Local
deploy-mode	是否发布你的驱动到 worker 节点（cluster）或者作为一个本地客户端 (client) (default: client)*
conf	任意的 Spark 配置属性， 格式 key=value. 如果值包含空格，可以加引号 “key=value”
application-jar	打包好的应用 jar，包含依赖，这个 URL 在集群中全局可见。 比如 hdfs:// 共享存储系统， 如果是 file:// path， 那么所有的节点的 path 都包含同样的 jar
application-arguments	传给 main() 方法的参数
executor-memory 1G	指定每个 executor 可用内存为 1 G
total-executor-cores 2	指定每个 executor 使用的 cup 核数为 2 个

2. 结果展示
   

2.1.4 WordCount 案例

1. 在 spark 目录下创建 input 目录

mkdir input

2. 在 input 目录下创建 2 个文件 1.txt 和 2.txt，并分别输入以下内容

hello Word
hello Scala

Hello Spark

3. 启动 spark-shell

spark-shell

4. 查看 spark 状态

   ① 开启另一个窗口

jps

    ② 浏览器输入 master:4040 查看程序运行

5. 运行 WordCount 程序

sc.textFile("/hadoop/spark-2.3.1/input").flatMap(_.split(" ")).map((_,1)).reduceByKey(_+_).collect

6. 查看结果

   ① 控制台
   
   ② web 页面
   

7. WordCount 案例分析

   ① 提交任务分析：
   
   ② 数据流分析：

   sc.textFile("/hadoop/spark-2.3.1/input").flatMap(.split(" ")).map((,1)).reduceByKey(+).collect

   textFile("/hadoop/spark-2.3.1/input")：读取本地 input 文件夹数据；
   flatMap(_.split(" "))：压平操作，按照空格分割符将一行数据映射成一个个单词；
   map((_,1))：对每一个元素操作，将单词映射为元组；
   reduceByKey(+)：按照key将值进行聚合，相加；
   collect：将数据收集到Driver端展示。
   

2.2 Standalone 模式

2.2.1 概述

  构建一个由 Master+Slave 构成的 Spark 集群，Spark 运行在集群中。

2.2.2 安装使用

1. 修改 spark 的配置文件（/hadoop/spark-2.3.1/conf 目录下）

   ① 修改配置文件名称

mv slaves.template slaves

mv spark-env.sh.template spark-env.sh

    ② 配置 slave 文件，添加 work 节点

master
slave1
slave2

    ③ 修改 spark-env.sh 文件

SPARK_MASTER_HOST=master
SPARK_MASTER_PORT=7077

2. 修改 sbin 目录下的 spark-config.sh

export JAVA_HOME=/usr/local/java/jdk1.8.0_151

3. 同步 spark 目录给其他机器

xsync spark-2.3.1/

4. 同步配置文件，并分别使配置文件生效

xsync /etc/profile

source /etc/profile

5. 启动 spark 集群

sbin/start-all.sh

6. 查看

   ① 查看 jps
   
   ② 网页查看：master:8080
   

2.2.3 官方求 PI 案例

1. 启动 spark 集群

sbin/start-all.sh

2. 输入命令

spark-submit \
--class org.apache.spark.examples.SparkPi \
--master spark://master:7077 \
--executor-memory 1G \
--total-executor-cores 2 \
/hadoop/spark-2.3.1/examples/jars/spark-examples_2.11-2.3.1.jar \
100

2.2.4 WordCount 案例

1. 启动 spark shell

spark-shell \
--master spark://master:7077 \
--executor-memory 1g \
--total-executor-cores 2

  参数：–master spark://master:7077指定要连接的集群的 master

2. 执行 WordCount 程序

sc.textFile("/hadoop/spark-2.3.1/input").flatMap(_.split(" ")).map((_,1)).reduceByKey(_+_).collect

3. 查看运行结果
   

2.2.5 配置历史服务器

1. 修改 spark-default.conf.template 名称

mv spark-defaults.conf.template spark-defaults.conf

2. 修改 spark-default.conf 文件，开启 Log：

spark.eventLog.enabled           true
spark.eventLog.dir               hdfs://hadoop102:9000/directory

3. 在 HDFS 上创建 /directory 目录

hadoop fs -mkdir /directory

4. 修改 spark-env.sh 文件，添加如下配置：

# - SPARK_HISTORY_OPTS, to set config properties only for the history server (e.g. "-Dx=y")
SPARK_HISTORY_OPTS="-Dspark.history.ui.port=18080 
-Dspark.history.retainedApplications=30 
-Dspark.history.fs.logDirectory=hdfs://master:9000/directory"

5. 同步配置文件

xsync spark-defaults.conf

xsync spark-env.sh

6. 启动历史服务

sbin/start-history-server.sh

7. 再次执行任务

spark-submit \
--class org.apache.spark.examples.SparkPi \
--master spark://master:7077 \
--executor-memory 1G \
--total-executor-cores 2 \
/hadoop/spark-2.3.1/examples/jars/spark-examples_2.11-2.3.1.jar \
100

8. 查看历史服务

   master：18080
   

2.2.6 HA 配置

1. 启动 Zookeeper
2. 修改 spark-env.sh 文件添加如下配置：

#SPARK_MASTER_HOST=hadoop102
#SPARK_MASTER_PORT=7077

SPARK_DAEMON_JAVA_OPTS="
-Dspark.deploy.recoveryMode=ZOOKEEPER 
-Dspark.deploy.zookeeper.url=master,slave1,slave2
-Dspark.deploy.zookeeper.dir=/spark"

3. 同步配置文件

xsync spark-env.sh

4. 在 master上启动全部节点

sbin/start-all.sh

5. 在 slave1 单独启动 master 节点

sbin/start-master.sh

6. spark HA 集群访问

spark-shell \
--master spark://master:7077,slave1:7077 \
--executor-memory 2g \
--total-executor-cores 2

2.3 Yarn 模式

2.3.1 概述

1. Spark 客户端直接连接 Yarn，不需要额外构建 Spark 集群。有 yarn-client 和 yarn-cluster 两种模式，主要区别在于：Driver 程序的运行节点。

   ① yarn-client

     Driver 程序运行在客户端，适用于交互、调试，希望立即看到 app 的输出。

   ② yarn-cluster

     Driver 程序运行在由 RM（ResourceManager）启动的 AP（APPMaster）适用于生产环境。
   
   

2. YARN 运行模式介绍：

2.3.2 安装使用

1. 修改 hadoop 配置文件 yarn-site.xml（/hadoop/hadoop-2.7.7/etc/hadoop/），添加如下内容：

<!--是否启动一个线程检查每个任务正使用的物理内存量，如果任务超出分配值，则直接将其杀掉，默认是true -->
<property>
   <name>yarn.nodemanager.pmem-check-enabled</name>
   <value>false</value>
</property>

<!--是否启动一个线程检查每个任务正使用的虚拟内存量，如果任务超出分配值，则直接将其杀掉，默认是true -->
<property>
  <name>yarn.nodemanager.vmem-check-enabled</name>
  <value>false</value>
</property>

2. 同步配置

xsync yarn-site.xml

3. 修改 spark-env.sh（/hadoop/hadoop-2.7.7/spark-2.3.1/conf/），添加如下配置：

# - YARN_CONF_DIR, to point Spark towards YARN configuration files when you use YARN
YARN_CONF_DIR=/hadoop/hadoop-2.7.7/etc/hadoop

4. 同步配置

xsync spark-env.sh 

2.3.3 官方求 PI 案例

1. 启动 Hadoop 集群（HDFS、YARN）
2. 输入命令

spark-submit \
--class org.apache.spark.examples.SparkPi \
--master yarn \
--deploy-mode client \
/hadoop/spark-2.3.1/examples/jars/spark-examples_2.11-2.3.1.jar \
100

2.3.4 WordCount 案例

1. 启动 spark shell

spark-shell --master yarn

  参数：–master spark://master:7077指定要连接的集群的 master

2. 执行 WordCount 程序

sc.textFile("/hadoop/spark-2.3.1/input").flatMap(_.split(" ")).map((_,1)).reduceByKey(_+_).collect

3. 查看运行结果
   

2.3.5 配置历史服务器

1. 修改配置文件 spark-defaults.conf

   添加如下内容：

spark.yarn.historyServer.address=master:18080
spark.history.ui.port=18080

2. 重启 spark 历史服务

sbin/stop-history-server.sh 

sbin/start-history-server.sh 

3 提交任务到 Yarn 执行

spark-submit \
--class org.apache.spark.examples.SparkPi \
--master yarn \
--deploy-mode client \
/hadoop/spark-2.3.1/examples/jars/spark-examples_2.11-2.3.1.jar \
100

6. Web 页面查看日志
   
   

2.4 三种模式比较

模式	Spark安装机器数	需启动的进程	所属者
Local	1	无	Spark
Standalone	3	Master 及 Worker	Spark
Yarn	1	Yarn 及 HDFS	Hadoop

3. 案例实操

3.1 IDEA 编写 WordCount 程序

1. 创建一个 Maven 项目，并导入以下依赖

<dependencies>
    <dependency>
        <groupId>org.apache.spark</groupId>
        <artifactId>spark-core_2.11</artifactId>
        <version>2.3.1</version>
    </dependency>
</dependencies>

<build>
    <finalName>WordCount</finalName>
    <plugins>
        <plugin>
            <groupId>net.alchim31.maven</groupId>
            <artifactId>scala-maven-plugin</artifactId>
            <version>3.2.2</version>
            <executions>
                <execution>
                    <goals>
                        <goal>compile</goal>
                        <goal>testCompile</goal>
                    </goals>
                </execution>
            </executions>
        </plugin>
    </plugins>
</build>

2. 创建 scala 源码目录
   

3. 创建 input 文件夹，并创建 word.txt

   

hello Scala
hello Spark
hello World
hello Hadoop
hello Scala

4. 编写代码

package spark

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

object WordCount {
    
  def main(args: Array[String]): Unit = {
    

    //1.创建SparkConf
    val conf = new SparkConf().setMaster("local[*]").setAppName("WordCount")

    //2.创建SparkContext
    val sc = new SparkContext(conf)

    //3.使用sc创建RDD并执行相应的transformation和action
//    sc.textFile("input").flatMap(_.split(" ")).map((_, 1)).reduceByKey(_+_).collect()

    // 读取文件，将文件的内容一行一行的读出来
    val lines = sc.textFile("spark/input")
    // 将一行一行的数据分解一个一个的单词
    val words = lines.flatMap(_.split(" "))
    // 将单词的数据进行结构的转换
    val wordToOne = words.map((_, 1))
    // 对转换结构后的数据进行分组聚合
    val wordToSum = wordToOne.reduceByKey(_ + _)
    // 将统计结果采集后打印到控制台
    val result = wordToSum.collect()
    result.foreach(println)

    //4.关闭连接
    sc.stop()
  }
}

5. 运行结果
   
6. 在 pom.xml 中添加打包插件，将程序打包

<plugin>
  <groupId>org.apache.maven.plugins</groupId>
  <artifactId>maven-assembly-plugin</artifactId>
  <version>3.0.0</version>
  <configuration>
      <archive>
          <manifest>
              <mainClass>WordCount</mainClass>
          </manifest>
      </archive>
      <descriptorRefs>
          <descriptorRef>jar-with-dependencies</descriptorRef>
      </descriptorRefs>
  </configuration>
  <executions>
      <execution>
          <id>make-assembly</id>
          <phase>package</phase>
          <goals>
              <goal>single</goal>
          </goals>
      </execution>
  </executions>
</plugin>