Spark入门

前言

本人并未从事Spark相关的工作，但由于项目需要使用了Spark将算法实现并行化，所以本篇博客更多的是一些简单、直白的Spark用法与优化。适合看本篇博客的人应该是与我一样由于课题需要临时使用Spark或者说出于兴趣探索Spark，这篇博客可以给予一个基础的介绍。如果你从事使用Spark编程的工作，那么我更建议你简单看一看本篇博客后前往官网或者找基本认可度较高的书籍去系统的学习，因为本篇博客不可避免的会出现——对内容的浅尝辄止、一些其他常用方法的缺失、甚至是一些理解上的偏差。

简介

Apache Spark is a fast and general engine for large-scale data processing.

Spark是UC Berkeley AMP lab (加州大学伯克利分校的AMP实验室)所开源的类Hadoop MapReduce的通用并行框架，它是为了实现大数据的快速计算而设计的。Spark拥有Hadoop MapReduce所具有的优点；但不同于MapReduce的是Job中间输出结果可以保存在内存中，从而不再需要读写HDFS，因此Spark能更好地适用于数据挖掘与机器学习等需要迭代的MapReduce的算法。

语言

Spark本身是由Scala编写的。Scala(http://www.scala-lang.org/ )是一门集成了函数式编程和面向对象编程特性的语言，同时它与Java一样编译成class文件后运行于JVM中，它支持非常多的Java模块。

为了方便数据科学家们使用Spark去验证算法，Spark同时提供了Python与R的编程接口。当然，Spark也支持Scala甚至Java，而且它们有比Python或R更好的计算性能。

配置依赖

Spark可以不依赖于任何第三方框架单独运行，这时其部集群模式为standalone模式。与此同时，Spark还支持Meos与Yarn模式，它们都依赖于Apache Hadoop框架。在未来，Spark还将支持Kubernetes平台。

关于三种模式的使用，具体可以前往官方网站(http://spark.apache.org/docs/...查阅。

部署

由于本人尝试在集群上安装部署Spark失败了，所以这里不打算介绍如何在集群上部署Spark，而且更多的情况下，大家主要工作是去使用集群工作，部署的事情交给运维们就好了。

工作方式

Spark既可以使用shell进行脚本编程来编写一些简单的代码熟悉接口或测试环境，也可以编写代码文件提交应用到集群运行。

你可以在terminal中输入spark-shell指令来启动scala的shell，也可以输入pyspark指令来启动python的shell。在启动shell时，将默认初始化SparkContext对象（SparkContext是所有Spark接口的入口，它包含一个与Spark集群的连接），你可以直接在脚本中使用sc变量来访问到它。如果你想要向SparkConf中添加其他设置（SparkConf是一个Spark应用的设置，它是SparkContext初始化的可选参数），你可以在输入启动指令的同时在后面添加形如--options value的指令来启动shell，具体设置参数的名字与取值可以前往 spark.apache.org/docs/latest/configuration.html 查询。

当运行结构复杂的代码时，你便需要使用spark-submit参数来提交你的应用。以python和scala为例，对于python代码编写的应用，你可以直接提交程序入口的py文件，但还需要添加--files来向Spark提交程序需要访问的本地文件（不同文件名之间用逗号隔开），以及添加--py-files来向Spark提交主文件需要调用到的你自己编写的模块（不同文件名之间用逗号隔开）；对于scala代码编写的应用，你需要使用工具（例如sbt）来打包你的scala代码与资源文件，然后直接提交jar文件即可。

RDD与DataFrame

撇开Spark中的DAG框架GraphX不谈，Spark中的两种分布式数据结构便是RDD与DataFrame。RDD与DataFrame都是模板类，其本身是一个集合，它们的集合元素会在计算时分发到不同的节点上。DataFrame是类似SQL中表格格式的存储，而RDD只是单纯对集合元素的封装。除非你的应用需要与SQL数据库交互，或者你的程序需要用到ml等仅支持DataFrame类型参数的模块，否则具备优秀灵活性的RDD毫无疑问是更好的选择。

Transformation与Action

在RDD数据上Spark支持Transformation与Action两种类型的操作。

常见的Transformation操作是map、flatmap、distinct等函数，它们在RDD上执行分布式数据的映射，转换前后的数据一一对应。注意，Spark中的所有Transformation函数都是lazy的，当你调用它们时，Spark仅做了映射规则以及变量的一些记录，仅当Transformation后的RDD变量执行Action操作时，之前的Transformation才会被执行，而这一Transformation的执行结果除去用于Action操作外不会被记录，换言之，对于类似下面的代码

mapData = rddData.map(func)
result1 = mapData.collect()
result2 = mapData.count()

对rdd执行func的amp会运行两次，而不是想象中的map结果被保存下来并用于两次Action计算。如果想避免这种情况，可以使用persist函数保留Transformation的结果。

常见的Action操作则包括collect、count等函数，Action函数返回的不再是分布式数据对象而是原生数据结构，如RDD[T]对象collect会返回Array[T]，而count会返回一个符合条件的集合元素数量的Int值。

node、worker与executor

由于涉及到重要的参数设置，这三个概念的关系必须要先理清。

node就是物理意义上集群里的一台机器，worker运行于node上，executor运行于worker中。worker可以调度的资源受集群启动时node设置的限制，executor可以调度的资源受应用启动时worker设置的限制。一个node上可以有多个worker，一个worker中可以有多个executor。一个应用使用的node数受集群启动时设置限制，无法少用或多用节点；一个应用总是使用每个node上的一个worker；一个应用总是使用每个worker中的所有executor。

executor是spark中任务执行的最小单元，manager会在一个stage中一次将分布式数据的一个切片分发给一个executor进行计算。当计算完“一片”数据后，若RDD还有数据切片未被分发到executor，则manager会将下一切片数据分发给计算完毕的executor。一个executor可以使用多个CPU核心，计算“一片”数据时会根据一个executor可以调用的CPU核心数进行多线程的并行计算。

那么假如独占集群，一个node仅有一个worker并可以调用node上所有的CPU核心、而一个worker仅有一个executor并可以调用所有的worker核心时，集群利用率应该是最高。

常用参数说明

在代码、提交指令的选项与环境配置文件中都可以对应用的Spark属性进行设置。假如对相同属性设置了不同的值，其优先级按上文顺序依次下降。

master

无论是使用SparkConf还是直接使用SparkContext来启动与集群的连接，master都是必须设置的一个参数，其默认值为"local[*]"，表示以伪集群模式按driver的物理核心数启动伪worker。如果仅是在一台电脑上想要体验Spark的并行计算，可以通过调整local中的参数值来设置worker的个数。如果想要使用到集群，必须将其设置为集群的master机IP"spark://xxxx:xxxx"。对于yarn模式直接设置为"yarn"即可。

driver memory

driver是应用的“起点”，代码、资源文件都被提交到driver，同时主函数中的变量与分布式数据Action的返回值都会存储于driver的内存。这一参数用于设置driver机器的可用内存上限，即使设置值超过机器的实际内存大小Spark也会进行内存交换来满足应用要求，但是如果应用在driver上使用的内存超过了设置值则spark不会进行内存交换而是会直接报错。注意，虽然spark文档上说道SparkConf可以设置spark.driver.memory，但实际上driver的启动先于提交的应用代码运行，所以如需设置请写在指令或配置文件中。

driver maxResultSize

上面提到，Action返回值会存储在driver中。这一参数用于设置Action返回值的数据大小上限，超过该值Spark应用会报错并终止运行。设置为0表示不设置上限。

executor instances

该参数用于设置一个worker中的executor数量。注意，该参数仅在yarn模式中生效，倘若出于测试加速比等原因需要在standalone模式中设置executor数量，可以使用cores max参数配合executor cores参数来间接设置。

executor cores

该参数用于设置一个executor所用的CPU核心数，启动的每个executor都必须满足这一设置，假如设置值大于worker cores或者由于在同一worker上已有其他executor导致核心数不满足条件，则executor不会启动。

cores max

应用可以使用的最大核心数。配合executor参数，可以启动(cores max)/(executor cores)个executor。

default parallelism

该参数用于设置分布式数据默认的切片数量或者说并行度，并行度本身也可以在parallelize等创建分布式数据的函数中作为参数传入。理论上，该值等于应用总核心数，即应用同一时间运行的最大线程数时，计算效率最高。但是由于并行度低时切片数据更大，容易出现超出内存空间而导致发生磁盘读写的情况。相比多次分发数据，磁盘读写是一个极为耗时的操作。所以实际中，应该通过估计或者实验，选择一个最佳的并行度，它应该是总核心数的整数倍。

编程

除了需要用你自己的数据集合初始化RDD或其他分布数据类、以及将对数据的操作封装成函数或写成lambda表达式并传入Transformation或Action的接口中外，Spark的编程与普通编程并没有什么不同。更多的工作在于了解各个接口的含义来完成并行化，以及了解Spark的底层机制来进行应用性能的优化。不过这是一篇入门文章，所以我还是放一段官方example代码来解释一下

conf = SparkConf().setAppName(appName).setMaster(master)
sc = SparkContext(conf=conf)
lines = sc.textFile("data.txt")
lineLengths = lines.map(lambda s: len(s))
totalLength = lineLengths.reduce(lambda a, b: a + b)

这一段代码足够简单，它调用了Transformation中的map与Action中的reduce，它们也是并行编程中最常用的操作之一。前两行代码里进行了Spark的初始化设置；第三行进行了对象初始化并返回RDD[String]类型的数据（当然python中不指定数据类型，这只是方便理解），textFile将文件中的数据按行读取并切片，它也支持传入第二个参数来设定切片数量；第四行中进行了map操作，创建了一个新的RDD对象存储文件中每一行的长度（记住之前说的，这里其实还没有真正执行map），同时这里说明一下RDD还是immutable的，即使真正执行了map也并不是将原RDD的数据进行了转换而是创建了新的RDD对象并存储原RDD数据映射的结果；最后一行里对RDD进行reduce操作，将每一个集合元素的值累加并返回，reduce的传入函数或lambda表达式要求有两个与返回值数据类型相同的参数，它将RDD中元素两两之间进行计算并作为下一次计算的传入参数，底层中可能是并行化操作所以假如n个数据只需要logn次计算的时间（没有去查资料验证，感兴趣自己去查MapReduce的论文）。

结语

Spark本身远不止这些应用，ml与mllib机器学习库、更高效的GraphX有向图框架、更多灵活方便的API，甚至业界主流的基于Spark的机器学习框架BigDL，都是有用的工具。希望这一篇博客可以帮助你快速上手Spark。如果你想要更深入地了解Spark，除了官网的program guide外可以帮助你了解API的用法与一些底层理论外，Jacek Laskowski的"Mastering Apache Spark 2"也可以帮助你了解Spark的框架结构，有兴趣的话可以去查阅。