大数据技术之Spark（一）——Spark概述

大数据技术之Spark（一）——Spark概述

前言

Apache Spark是一个开源的、强大的分布式查询和处理引擎，它提供MapReduce的灵活性和可扩展性，但速度明显要快上很多；拿数据存储在内存中的时候来说，它比Apache Hadoop 快100倍，访问磁盘时也要快上10倍。

一、Spark基础

1.1 Spark是什么

Spark 是一种由 Scala 语言开发的快速、通用、可扩展的大数据分析引擎。

Spark Core：Spark Core包含Spark的基本功能，如内存计算、任务调度、部署模式、故障恢复、存储管理等。Spark建立在统一的抽象RDD之上，使其可以以基本一致的方式应对不同的大数据处理场景；通常所说的Apache Spark，就是指Spark Core；

Spark SQL：兼容HIVE数据，提供比Hive更快的查询速度（10~100x）的分布式SQL引擎，开发者可以轻松地使用SQL命令进行查询，并进行更复杂的数据分析；

Spark Streaming：流式计算分解成一系列小的批处理作业利用spark轻量级低时延的框架来支持流数据处理，目前已经支持Kafka，Flume等；

MLilb：提供基于Spark的机器学习算法库，包括聚类、分类、回归、协同过滤等，降低了机器学习的门槛，开发人员只要具备一定的理论知识就能进行机器学习的工作；
GraphX：提供图形计算框架，与Pregel/GraphLab兼容。

1.2 Spark VS Hadoop

尽管 Spark 相对于 Hadoop 而言具有较大优势，但 Spark 并不能完全替代 Hadoop，Spark 主要用于替代Hadoop中的 MapReduce 计算模型。存储依然可以使用 HDFS，但是中间结果可以存放在内存中；调度可以使用 Spark 内置的，也可以使用更成熟的调度系统 YARN 等。

Spark 和 Hadoop 的根本差异是多个作业之间的数据通信问题 ；Spark 多个作业之间数据通信是基于内存，而 Hadoop 是基于磁盘。
实际上，Spark 已经很好地融入了 Hadoop 生态圈，并成为其中的重要一员，它可以借助于 YARN 实现资源调度管理，借助于 HDFS 实现分布式存储。

此外，Hadoop 可以使用廉价的、异构的机器来做分布式存储与计算，但是，Spark 对硬件的要求稍高一些，对内存与 CPU 有一定的要求。

1.3 Spark优势及特点

1.3.1 优秀的数据模型和丰富计算抽象

首先看看MapReduce，它提供了对数据访问和计算的抽象，但是对于数据的复用就是简单的将中间数据写到一个稳定的文件系统中(例如 HDFS)，所以会产生数据的复制备份，磁盘的I/O以及数据的序列化，所以在遇到需要在多个计算之间复用中间结果的操作时效率就会非常的低。而这类操作是非常常见的，例如迭代式计算，交互式数据挖掘，图计算等。

因此 AMPLab 提出了一个新的模型，叫做 RDD。

• RDD 是一个可以容错且并行的数据结构（其实可以理解成分布式的集合，操作起来和操作本地集合一样简单)，它可以让用户显式的将中间结果数据集保存在 内存 中，并且通过控制数据集的分区来达到数据存放处理最优化。同时 RDD 也提供了丰富的 API (map、reduce、filter、foreach、redeceByKey...)来操作数据集。

后来 RDD 被 AMPLab 在一个叫做 Spark 的框架中提供并开源。

1.3.2 spark的特点

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

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

通用：Spark 提供了统一的解决方案。Spark 可以用于批处理、交互式查询(Spark SQL)、实时流处理(Spark Streaming)、机器学习(Spark MLlib)和图计算(GraphX)，这些不同类型的处理都可以在同一个应用中无缝使用。

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

1.4 Spark 运行环境

① local 本地模式(单机) - 不需要其他任何节点资源就可以在本地执行Spark代码的环境

• 学习测试使用。
• 分为 local 单线程和 local-cluster 多线程。

② standalone 独立集群模式

• 学习测试使用。
• 典型的 Mater/slave 模式。

③ standalone-HA 高可用模式

• 生产环境使用
• 基于 standalone 模式，使用 zk 搭建高可用，避免 Master 是有单点故障的。

④ on yarn 集群模式

• 生产环境使用
• 运行在 yarn 集群之上，由 yarn 负责资源管理，Spark 负责任务调度和计算。
• 好处：计算资源按需伸缩，集群利用率高，共享底层存储，避免数据跨集群迁移。

⑤ on mesos 集群模式

• 国内使用较少
• 运行在 mesos 资源管理器框架之上，由 mesos 负责资源管理，Spark 负责任务调度和计算。

⑥ on cloud 集群模式

• 中小公司未来会更多的使用云服务
• 比如 AWS 的 EC2，使用这个模式能很方便的访问 Amazon 的 S3。

1.5 Spark运行架构

Spark 框架的核心是一个计算引擎，整体来说，它采用了标准 master-slave 的结构。
如下图所示，它展示了一个 Spark 执行时的基本结构。图形中的 Driver 表示 master，负责管理整个集群中的作业任务调度。图形中的 Executor 则是 slave，负责实际执行任务。

1、一个行动算子对应一个job
2、job中分为多个stage，通过是否shuffle来判定。shuffle一次一个stage（DAG有向无环图）
3、最后task的数量由rdd算子的分区数量来决定。rdd算子定义了几个分区，就有几个task

1.5.1 Driver

Spark驱动器节点，用于执行Spark任务中的main方法，负责实际代码的执行工作。Driver在Spark作业执行时主要负责：

• 将用户程序转化为作业（job）
• 在Executor之间调度任务(task)
• 跟踪Executor的执行情况
• 通过UI展示查询运行情况

1.5.2 Executor

Spark Executor是集群中工作节点（Worker）中的一个JVM进程，负责在Spark 作业中运行具体任务（Task），任务彼此之间相互独时启动，并且始终伴随着整个Spark 应用的生命周期而存在。如果有Executor节点发生了故障或崩溃，Spark 应用也可以继续执行，会将出错节点上的任务调度到其他Executor节点上继续运行。
Executor有两个核心功能：

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

1.5.3 Master & Worker

Spark集群的独立部署环境中，不需要依赖其他的资源调度框架，自身就实现了资源调度的功能，所以环境中还有其他两个核心组件：Master和Worker，这里的Master是一个进程，主要负责资源的调度和分配，并进行集群的监控等职责，类似于Yarn环境中的RM, 而Worker呢，也是进程，一个Worker运行在集群中的一台服务器上，由Master分配资源对数据进行并行的处理和计算，类似于Yarn环境中NM。

1.5.4 ApplicationMaster

Hadoop用户向YARN集群提交应用程序时,提交程序中应该包含ApplicationMaster，用于向资源调度器申请执行任务的资源容器Container，运行用户自己的程序任务job，监控整个任务的执行，跟踪整个任务的状态，处理任务失败等异常情况。
说的简单点就是，ResourceManager（资源）和Driver（计算）之间的解耦合靠的就是ApplicationMaster。

1.6 核心概念

1.6.1 Executor与Core

Spark Executor是集群中运行在工作节点（Worker）中的一个JVM进程，是整个集群中的专门用于计算的节点。在提交应用中，可以提供参数指定计算节点的个数，以及对应的资源。这里的资源一般指的是工作节点Executor的内存大小和使用的虚拟CPU核（Core）数量。
应用程序相关启动参数如下：

1.6.2 并行度（Parallelism）

在分布式计算框架中一般都是多个任务同时执行，由于任务分布在不同的计算节点进行计算，所以能够真正地实现多任务并行执行，记住，这里是并行，而不是并发。这里我们将整个集群并行执行任务的数量称之为并行度。那么一个作业到底并行度是多少呢？这个取决于框架的默认配置。应用程序也可以在运行过程中动态修改。

1.6.3 有向无环图（DAG）

这里所谓的有向无环图，并不是真正意义的图形，而是由Spark程序直接映射成的数据流的高级抽象模型。简单理解就是将整个程序计算的执行过程用图形表示出来,这样更直观，更便于理解，可以用于表示程序的拓扑结构。
DAG（Directed Acyclic Graph）有向无环图是由点和线组成的拓扑图形，该图形具有方向，不会闭环。

sc.parallelize(1 to 10).map((_,1)).reduceByKey(_+_)

工作原理

• 根据RDD之间的依赖关系，形成一个DAG DAG
• Scheduler将DAG划分为多个Stage

划分依据：是否发生宽依赖（shuffle）
划分规则：从后往前，遇到宽依赖且各位新的stage
每个stage由一组并行的Task组成

宽依赖与窄依赖

二、安装方式

Linux中Spark的安装方式

输入spark-shell运行，显示Spark context(sc)和Spark session(spark)。
SparkContext和SparkSession的区别

WordCount方法实现

// 1. 读取文件，获取一行一行的数据
// hello world
val lines = sc.textFile("datas")

// 2. 将一行数据进行拆分，形成一个一个的单词（分词）
// "hello world" => hello, world, hello, world
val words = lines.flatMap(_.split(" "))

// 3. 将数据根据单词进行分组，便于统计
// (hello, hello, hello), (world, world)
val wordGroup = words.groupBy(word => word)

// 4. 对分组后的数据进行转换
//  (hello, hello, hello), (world, world)
// (hello, 3), (world, 3)
val wordToCount = wordGroup.map {
	case(word, list) => {
		(word, list.size)
	}
}

// 5. 将转换结果采集到控制台打印出来
wordToCount.collect.foreach(println)

Java中SparkContext的创建与配置

object WordCount {
  def main(args: Array[String]): Unit = {
// Spark配置
    val conf = new SparkConf().setMaster("local[*]").setAppName("wordcount")
// 创建SparkContext
    val sc = SparkContext.getOrCreate(conf)

// 从hdfs中读取文件
    val rdd = sc.textFile("hdfs://192.168.153.139:9000/tmp/words.txt")
    val result = rdd.flatMap(x=>x.split("\t")).map((x=>(x,1))).reduceByKey(_+_)
// 保存文件并输出
    result.saveAsTextFile("hdfs://192.168.153.139:9000/tmp/wordsResult.txt")
  }

}

编译jar包

程序编写完成后，将生成的jar包放到linux中。这里我放在了/opt目录下，尝试编译代码。由于输入输出固定，我们可以直接在9870端口中查看是否成功输出名为wordResult.txt的文件

[root@hadoop02 opt]# spark-submit --class org.example.WordCount --master local[*] ./sparkstu-1.0-SNAPSHOT.jar 

这样就成功啦~

三、Spark核心编程

Spark计算框架为了能够进行 高并发 和 高吞吐 的数据处理，封装了三大数据结构，用于处理不同的应用场景。三大数据结构分别是：

• RDD : 弹性分布式数据集
• 累加器：分布式共享 只写 变量
• 广播变量：分布式共享 只读 变量

3.1 RDD

RDD（Resilient Distributed Dataset）叫做弹性分布式数据集，是Spark中最基本的 数据处理模型 。代码中是一个抽象类，它代表一个弹性的、不可变、可分区、里面的元素可并行计算的集合。

Resilient ：它是弹性的，RDD 里面的中的数据可以保存在 内存 中或者 磁盘 里面。

存储的弹性：内存与磁盘的自动切换
容错的弹性：数据丢失可以自动回复
计算的弹性：计算出错重试机制
分片的弹性：可根据需要重新分片

Distributed ： 它里面的元素是分布式存储的，可以用于分布式计算。数据存储在大数据集群不同节点上。

Dataset: 它是一个集合，封装了计算逻辑，并不保存数据。

3.1.1 RDD创建：parallelize和makeRDD

在spark中创建RDD的创建方式可以分为四种：

1）从集合（内存）中创建RDD

从集合中创建RDD，Spark主要提供了两个方法：parallelize和makeRDD

parallelize：

scala> var rdd = sc.parallelize(1 to 4)
rdd: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[168] at parallelize at <console>:24
scala> rdd.collect
res107: Array[Int] = Array(1, 2, 3, 4)

makeRDD：
从底层代码实现来讲，makeRDD方法其实就是parallelize方法

scala> var rdd = sc.parallelize(1 to 4)
rdd: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[168] at parallelize at <console>:24
scala> rdd.collect
res107: Array[Int] = Array(1, 2, 3, 4)
```

2）从外部存储（文件）创建RDD

由外部存储系统的数据集创建RDD包括：本地的文件系统，所有Hadoop支持的数据集，比如HDFS、HBase等。

// 从文件中创建RDD，将文件中的数据作为处理的数据源
// 1. path路径默认以当前环境的根路径为基准。可以写绝对路径，也可以写相对路径。
// 2. path路径可以是文件的具体路径，也可以是目录名称
// 3. path路径还可以使用通配符 *
// 4. path路径可以是分布式存储系统路径HDFS
scala> val rdd = sc.textFile("hdfs://hadoop02:9000/tmp/wordcount.txt")
scala> val rdd = sc.textFile("file:///opt/stufile/words.txt")

textFile：以行为单位来读取数据。读取的数据都是字符串
wholeTextFiles：以文件为单位读取数据。读取的结果表示为元组。第一个元素表示文件路径，第二个元素表示文件内容

3）从其他RDD创建

通过一个RDD运算完后，再产生新的RDD。

4）直接创建RDD（new）

使用new的方式直接构造RDD，一般由Spark框架自身使用。

3.1.2 RDD并行度与分区

默认情况下，Spark可以将一个作业切分多个任务后，发送给Executor节点并行计算，而能够并行计算的任务数量我们称之为并行度。
需要注意的是，并行执行的任务数量 ≠ 切分任务的数量。

// RDD的并行度 & 分区
// makeRDD方法可以传递第二个参数，用于表示分区的数量
// makeRDD第二个参数可以不传递，会使用默认值
val rdd = sc.makeRDD( List(1,2,3,4), 2 )

// 将处理的数据保存成分区文件
rdd.saveAsTextFile("hdfs://hadoop02:9000/tmp/rdd2_output")

3.1.3 RDD算子

我们放在下一篇里面详细介绍=>RDD算子

3.1.4 RDD序列化

RDD为什么需要序列化，实现序列化的方法

3.1.5 RDD依赖关系

RDD依赖关系：宽依赖、窄依赖

3.1.6 RDD持久化

RDD持久化：cache/persisit缓存，checkpoint检查点

3.1.7 RDD分区器Partitioner

​val rdd = sc.makeRDD(lIST(
    ("AAA","1231231234"),
    ("BBB","12341234123"),
    ("CCC","1212123"),
    ("DDD","111222333")
),3)

val partRDD = rdd.partitionBy(new MyPartitioner)
/**
  * 自定义分区器
  * 1. 继承Partitioner
  * 2. 重写方法
  */
    class MyPartitioner extends Partitioner {
        // 分区数量
        override def numPatririons: Int = 3

        // 根据数据的key值，返回数据所在的分区索引（从0开始）
        override def getPartition(key: Any): Int = {
            key match {
                case "AAA" => 0
                case "BBB" => 1
                case _ => 2
            }
        }
    }

3.1.8RDD文件读取与保存

Spark的数据读取及数据保存可以从两个维度来作区分：文件格式 和 文件系统。

文件格式：text文件、csv文件、sequence文件以及Object文件
文件系统：本地文件系统、HDFS、HBASE以及数据库

text文件：
读取：sc.textFile("input/1.txt")
保存：rdd.saveAsTextFile("output1");

sequence文件：
读取：sc.sequenceFile[Int,Int]("output").collect().foreach(println)
保存：rdd.saveAsSequenceFile("output2");

object文件：
读取：sc.objectFile[Int]("output").collect().foreach(println)
保存：rdd.saveAsObjectFile("output3");

3.2 累加器

举例说明：实现数字的累加运算

object AccDemo {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setMaster("local[*]").setAppName("checkPoint")
    val sc = new SparkContext(conf)

	val rdd = sc.makeRDD(List(1,2,3,4))
// 方法一：reduce 既有分区内计算，又有分区间计算
	val i = rdd.reduce(_+_)
	println(i)
// 方法二：使用foreach实现累加
	var sum = 0;
	rdd.foreach(
	  num => { sum += num }
	)
	println("sum = " + sum)

	sc.stop()
  }
}

出来的结果明显与期望不符：

——为什么结果不是10？
foreach不是分布式的，但是spark是分布式的。
也就意味着每个分区都会执行foreach，导致结果不确定。
——为什么结果是0？
1. sum在foreach之外，所以sum在Driver端进行初始化，sum=0
2. sum在foreach中执行相加操作
3. 由于是分布式执行，所以两个Executor都会执行sum+=num的操作
4. 由于存在闭包检测，所以sum=0被传到Executor端
5. 计算完毕之后，应该将sum计算的返回结果给Driver。但这里不存在返回操作。
6. 所以说，执行计算的是Executor端的sum，对Driver端的sum不造成影响。

那当我们想要实现累加的功能，但闭包的计算结果无法返回到Driver端，此时需要一种特殊结构，可以将Driver端的数据传给Executor，当计算结束之后，可以将结果返回给Driver，类似于如下图：

总结：
累加器用来把 Executor 端变量信息聚合到 Driver 端。
在 Driver 程序中定义的变量，在Executor 端的每个 Task 都会得到这个变量的一份新的副本，每个 task 更新这些副本的值后，传回 Driver 端进行 merge。

累加器的实现

object acc {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setMaster("local[*]").setAppName("checkPoint")
    val sc = new SparkContext(conf)

    val rdd = sc.makeRDD(List(1,2,3,4))
    
    // 获取系统累加器
    // spark默认提供了简单的数据聚合的累加器
    var sumAcc = sc.longAccumulator("sum")
	rdd.foreach(
	  // 使用累加器
      num => { sumAcc.add(num) }
    )
	// 获取累加器的值
	println(sumAcc.value)
    

    sc.stop()
  }
}

注意：一般情况下，累加器会放置在行动算子中进行操作。

3.3 广播变量

3.3.1 实现原理

广播变量用来高效分发较大的对象。向所有工作节点发送一个较大的只读值，以供一个或多个 Spark 操作使用。比如，如果你的应用需要向所有节点发送一个较大的只读查询表，广播变量用起来都很顺手。在多个并行操作中使用同一个变量，但是 Spark 会为每个任务分别发送。文章来源地址https://www.toymoban.com/news/detail-622903.html

3.3.2 基础编程

object BraodCastDemo {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setAppName("BroadCast").setMaster("local[*]")
    val sc = new SparkContext(conf)

// 声明广播变量
    val arr = Array("hello","hi","com on baby")
    val broadcastVar = sc.broadcast(arr)
    
    val rdd = sc.parallelize(Array((1,"leader"),(2,"teamLeader"),(3,"worker")))
    val rdd2 = rdd.mapValues(x=>{
      println("value is" + x)
      
      // 使用广播变量
      broadcastVar.value(2) + ":" + x
    })
    rdd2.collect.foreach(println)

	sc.stop()
  }
}

到了这里，关于大数据技术之Spark（一）——Spark概述的文章就介绍完了。如果您还想了解更多内容，请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章，希望大家以后多多支持TOY模板网！