下载掌阅APP，畅读海量书库
立即打开
本节主要介绍Spark框架背景及其基本概念。
Spark是加州大学伯克利分校的AMP实验室（Algorithms Machines and People Lab）开源的类Hadoop MapReduce的通用并行框架，拥有Hadoop MapReduce所具有的优点。但它不同于MapReduce的是，其job（作业）中间输出的结果可以保存在内存中，从而不再需要读写HDFS。因此，Spark能更好地适用于数据挖掘与机器学习等需要迭代的MapReduce的算法。
Spark是一种与Hadoop相似的开源集群计算环境，但是两者之间存在一些不同之处，这些不同之处使Spark在处理某些工作负载方面表现得更加优越。换句话说，Spark启用了内存分布数据集，除了能够提供交互式查询外，还可以优化迭代工作负载。
Spark是使用Scala语言实现的，它将Scala用作应用程序框架。因此，Spark和Scala能够紧密集成，Scala可以像操作本地集合对象一样轻松地操作分布式数据集。
尽管创建Spark是为了支持分布式数据集上的迭代作业，但实际上它是对Hadoop的补充，可以在Hadoop文件系统中并行运行。通过名为Mesos的第三方集群框架可以支持此行为。Spark可以用来构建大型的、低延迟的数据分析应用程序。
Spark具有以下特性：
●高可伸缩性。Spark能够高效地处理从单一服务器到数千个节点的大规模集群上的数据。它的设计允许在多种集群管理器上运行，如Hadoop YARN、Apache Mesos或Kubernetes，并且能够灵活地扩展资源以满足不同的数据处理需求。
●高容错。Spark采用基于数据分片的容错机制，如数据复制和RDDs的不可变特性，确保了在节点故障时数据的完整性和计算任务的可靠性。
●内存计算。Spark的独特优势之一是其内存计算能力。通过将数据缓存到内存中，Spark能够实现比传统基于磁盘的系统更快的处理速度。这种内存计算特性显著提高了迭代算法和交互式数据查询的性能。
这些特性使得Spark成为处理大规模数据集和实现复杂数据分析任务的理想选择。
Spark属于BDAS（伯利克分析栈）生态体系。
●MapReduce属于Hadoop生态体系之一，Spark则属于BDAS生态体系之一。
●Hadoop包含了MapReduce、HDFS、HBase、Hive、ZooKeeper、Pig、Sqoop等。
●BDAS包含了Spark GraphX、Spark SQL（相当于Hive）、Spark MLlib、Spark Streaming（消息实时处理框架，类似Storm）、BlinkDB等。
BDAS生态体系如图2-1所示。
相对于MapReduce，Spark具有以下优势：
●MapReduce通常将中间结果存放到HDFS上；Spark则是基于内存并行大数据框架，中间结果存放到内存。对于迭代数据而言，Spark的效率更高。
●MapReduce总是消耗大量时间排序，而有些场景不需要排序；Spark则可以避免不必要的排序所带来的开销。
图2-1 BDAS生态体系
●Spark是一张有向无环图（从一个点出发最终无法回到该点的一个拓扑），并对有向无环图对应的流程进行优化。
Spark为什么比MapReduce快？简单地说，有以下3点原因：
（1）Spark基于内存计算，减少了低效的磁盘交互。
（2）Spark使用基于DAG（Directed Acyclic Graph，有向无环图）的高效调度算法。
（3）Spark具有容错机制Linage（血统）。
Spark支持的API包括Scala、SQL、Python、Java、R等。
Spark有以下5种运行模式，其中Local是单机模式，其他4种都是集群模式。
●Local：Spark运行在本地模式上，用于测试、开发。本地模式就是使用一个独立的进程，通过其内部的多个线程来模拟整个Spark运行时环境。
●Standalone：Spark运行在独立集群模式上。Spark中的各个角色以独立进程的形式存在，并组成Spark集群环境。
●Hadoop YARN：Spark运行在YARN上。Spark中的各个角色运行在YARN的容器内部，并组成Spark集群环境。
●Apache Mesos：Spark中的各个角色运行在Apache Mesos上，并组成Spark集群环境。
●Kubernetes：Spark中的各个角色运行在Kubernetes的容器内部，并组成Spark集群环境。
本节简单讲解一下Spark集群的相关知识。
Spark应用程序（application）在集群上作为独立的进程集运行，由驱动程序（driver program，又称为主程序）中的SparkContext对象进行协调。具体来说，Spark应用程序要在集群上运行，SparkContext可以连接到几种类型的集群管理器（cluster manager，包括Spark自己的独立集群管理器、Mesos、YARN或Kubernetes），这些集群管理器在应用程序之间分配资源；连接后，Spark会获取集群中工作节点（worker node）上的执行器（executor），这些节点是为应用程序运行计算和存储数据的进程；接下来，它将应用程序代码（由传递给SparkContext的JAR或Python文件定义）发送给执行器；最后，SparkContext将任务发送给执行器来运行，如图2-2所示。
图2-2 Spark执行步骤
驱动程序启动多个工作节点，这些工作节点从文件系统加载数据并产生RDD（即将数据存放到RDD中，RDD是一个数据结构），再按照不同分区缓存到内存中。
这个集群架构中有几个要点需要注意：
（1）每个应用程序都有自己的执行器进程，这些进程在整个应用程序期间保持运行，并在多个线程中运行任务。这样做的好处是在调度端（每个驱动程序调度自己的任务）和执行器端（来自不同应用程序的任务在不同JVM中运行）将应用程序彼此隔离。然而，这也意味着，如果不将数据写入外部存储系统，就无法在不同的Spark应用程序（SparkContext实例）之间共享数据。
（2）Spark对底层集群管理器是不可知的。只要它可以获取执行器进程，并且这些进程相互通信，那么就可以在其他应用程序（例如Mesos/YARN/Kubernetes）的集群管理器上运行。
（3）驱动程序必须在其整个生命周期中监听并接收来自执行程序的传入连接（例如，网络配置部分中的spark.driver.port）。因此，驱动程序必须是可从工作节点进行网络寻址的。
（4）因为驱动程序在集群上调度任务，所以它应该在工作节点附近运行，最好在同一局域网上运行。如果想远程向集群发送请求，最好打开一个RPC（remote procedure call，远程进程调度）到驱动程序，让它从附近节点提交操作（指transformation和action），而不是在远离工作节点的地方运行驱动程序。
Spark当前支持以下几种集群管理器：
●Standalone：Spark附带的一个简单的集群管理器，可以轻松地设置集群。
●Apache Mesos：一个通用的集群管理器，也可以运行Hadoop MapReduce和服务应用程序。（已弃用）
●Hadoop YARN：Hadoop 2和Hadoop 3中的资源管理器。
●Kubernetes：一个用于自动化和容器化应用程序的部署、扩展和管理的开源系统。
Spark可以控制应用程序之间（在集群管理器级别）和应用程序内部（如果在同一SparkContext上进行多个计算）的资源分配。
Spark集群的常用术语如表2-1所示。
表2-1 Spark集群常用术语
RDD英文名为Resilient Distributed Dataset，中文名为弹性分布式数据集。
什么是RDD？RDD是一个只读、分区记录的集合，可以把它理解为一个存储数据的数据结构。也就是说，RDD是Spark对数据的核心抽象，其实就是分布式的元素集合。在Spark中，对数据的所有操作不外乎创建RDD、转化已有RDD以及调用RDD操作进行求值。而在这一切操作背后，Spark会自动将RDD中的数据分发到集群上，并将操作并行化执行。在Spark中的一切操作都是基于RDD的。
RDD可以通过以下3种方式创建：
●集合转换。
●从文件系统（本地文件、HDFS、HBase）输入。
●从父RDD转换（为什么需要父RDD呢？为了容错）。
RDD的计算类型有以下两种：
●transformation：延迟执行。一个RDD通过该操作产生新的RDD时不会立即执行，只有等到action操作才会真正执行。
●action：提交Spark作业。当执行action时，transformation类型的操作才会真正执行计算操作，然后产生最终结果并输出。
Hadoop提供的处理数据的接口有Map和Reduce，而Spark提供的不仅有Map和Reduce，还有更多别的数据处理接口。Spark算子包括转换算子和行动算子，这部分内容将在3.4节集中讨论。
每个RDD都会记录自己所依赖的父RDD，一旦出现某个RDD的某些分区（partition）丢失，就可以通过并行计算迅速恢复，这就是容错。
RDD的依赖又分为窄依赖（narrow dependent）和宽依赖（wide dependent）。
（1）窄依赖：每个分区最多只能给一个RDD使用。由于没有多重依赖，因此在一个节点上可以一次性将分区处理完，并且一旦数据发生丢失或者损坏，可以迅速从上一个RDD恢复。
（2）宽依赖：每个分区可以给多个RDD使用。由于有多重依赖，因此只有等到所有到达节点的数据处理完毕才能进行下一步处理，一旦发生数据丢失或者损坏，需要从所有父RDD重新计算。相对窄依赖而言，宽依赖付出的代价更高。因此，在发生数据丢失或损坏之前，必须对上一次所有节点的数据进行物化（存储到磁盘上）处理，以便恢复，同时也应尽量减少宽依赖的使用。
RDD的宽依赖和窄依赖如图2-3所示。
图2-3 RDD的宽依赖和窄依赖
Spark提供了多种缓存策略，通过配置不同的参数组合实现。这些策略主要用于控制RDD数据的存储方式，以优化性能和资源使用。Spark的缓存策略由以下5个参数控制：
●useDisk：是否使用磁盘缓存。
●useMemory：是否使用内存缓存。
●useOffHeap：是否使用Java的堆外内存。
●deserialized：数据是否以反序列化的形式存储（序列化是为了方便数据在网络中以对象的形式进行传输）。
●replication：数据的副本数量。
Spark通过这5个参数组成11种缓存策略。
（1）DISK_ONLY：仅使用磁盘存储。
参数：_useDisk, _useMemory, _useOffHeap, _deserialized, _replication（默认值为1）。
（2）DISK_ONLY_2：使用磁盘存储，并进行2次数据副本复制。
参数：_useDisk, _useMemory, _useOffHeap, _deserialized, _replication（默认值为1）。
（3）MEMORY_ONLY：仅使用内存存储（默认策略）。
参数：_useDisk, _useMemory, _useOffHeap, _deserialized, _replication（默认值为1）。
（4）MEMORY_ONLY_2：使用内存存储，并进行2次数据副本复制。
参数：_useDisk, _useMemory, _useOffHeap, _deserialized, _replication（默认值为1）。
（5）MEMORY_ONLY_SER：使用内存存储，数据为序列化的形式，这可能会消耗更多的CPU资源。
参数：_useDisk, _useMemory, _useOffHeap, _deserialized, _replication（默认值为1）。
（6）MEMORY_ONLY_SER_2：结合MEMORY_ONLY_SER和2次数据副本复制。
参数：_useDisk, _useMemory, _useOffHeap, _deserialized, _replication（默认值为1）。
（7）MEMORY_AND_DISK：使用内存和磁盘存储，如果内存不足以存储所有数据，多余的部分将存储在本地磁盘上。
参数：_useDisk, _useMemory, _useOffHeap, _deserialized, _replication（默认值为1）。
（8）MEMORY_AND_DISK_2：结合MEMORY_AND_DISK和2次数据副本复制。
参数：_useDisk, _useMemory, _useOffHeap, _deserialized, _replication（默认值为1）。
（9）MEMORY_AND_DISK_SER：使用内存和磁盘存储，数据为序列化形式。
参数：_useDisk, _useMemory, _useOffHeap, _deserialized, _replication（默认值为1）。
（10）MEMORY_AND_DISK_SER_2：结合MEMORY_AND_DISK_SER和2次数据副本复制。
参数：_useDisk, _useMemory, _useOffHeap, _deserialized, _replication（默认值为1）。
（11）OFF_HEAP：使用堆外内存存储，不使用JVM堆内存，例如可以使用Tachyon作为堆外存储。
参数：_useDisk, _useMemory, _useOffHeap, _deserialized, _replication（默认值为1）、NONE。
NONE表示不需要缓存。
缓存策略通过StorageLevel类的构造传参的方式进行控制，结构如下：
class StorageLevel private(useDisk : Boolean ,useMemory : Boolean ,deserialized : Boolean ,replication：Ini)
Spark集群任务提交方式有3种：
●spark-submit（官方推荐）。
●sbt run。
●java -jar。
任务提交时可以指定各种参数，例如：
./bin/spark-submit
-- class <main- class >
--master <master-url>
--deploy-mode <deploy-mode>
--conf <key> = <value>
... # 其他选项
<application-jar>
[application-arguments]
spark-submit提交方式如下：
# 在本地模式下使用8个核心运行应用程序
./bin/spark-submit \
--class org.apache.spark.examples.SparkPi \
--master local[8]\
/path/to/examples.jar \
100
# 在客户端部署模式下运行Spark独立集群
./bin/spark-submit \
--class org.apache.spark.examples.SparkPi \
--master spark://207.184.161.138:7077
--executor-memory 20G \
--total-executor-cores 100 \
/path/to/examples.jar \
1000
每个驱动程序都有一个Web UI，通常位于端口4040上，用于显示有关正在运行的任务、执行器和存储使用情况的信息。在Web浏览器中访问http://<driver node>:4040即可访问此UI。