《Spark -- The Definitive Guide》学习笔记Note.3

上一篇分享，我们了解了一些关于Spark架构，应用程序，结构化API，核心概念的内容，有了这些作为基础，这期我们将对Spark工具集的相关内容进行译文和学习，分享过程中若有错谬，欢迎拍砖。

Charpter 3.A Tour of Spark’s Toolset

在第一期分享中，有下述图例，事实上，除了低级API和结构化API，Spark还提供了一系列标准库来实现额外功能，比如图分析，机器学习，流处理，以及其他计算和存储系统的集成能力等。

本期分享的主要内容有：

1、spark-submit；
2、dataset；
3、结构化流处理；
4、机器学习和高级分析；
5、RDD；
6、SparkR；
7、第三方软件包生态系统。

Running Production Applications

Spark让开发和创建大数据程序变得容易，其内置命令行工具 spark-submit，让用户可以将交互式开发程序转变为生产应用程序。spark-submit首先将应用程序发送到集群，然后启动它并在集群中执行。一旦提交，应用程序就会运行至结束或抛出报错。spark-submit支持在所有集群管理器（Standalone，Mesos和YARN）中执行。spark-submit 提供了一些控制选项，用户可以指定applications所需资源，运行应用程序运行方式，命令行参数。用户可以用Spark支持的所有语言编写程序并提交执行，最简单的就是在本地计算机上运行application。
现在我们运行一个Spark的example，它的功能是按一定精度计算pi值，local参数值表示本地运行，通过更改master参数也可以将应用提交给运行着Spark独立集群管理器，Mesos或YARN的集群，10是目标jar包的程序入参。

Datasets：Type-Safe Structured APIs

我们要说的第一个Spark结构化API是Dataset，它是类型安全版本，被用在Java和Scala中编写静态类型代码。它在Python和R中不可用，因为这些语言是动态类型的。上一篇博文分享的DataFrame是Row类型对象的分布式集合，被用来存储各种类型的表格数据。而Dataset API让用户可以用Java / Scala类，去定义DataFrame的每条记录，并将其作为类型对象集合进行操作，如Java的ArrayList，Scala的Seq。 Dataset上可用的API是类型安全的，即Dataset对象不会被视为与初始定义类不相同的另一个类。这在编写大型应用程序时十分有效，不同开发者可通过定义好的接口进行交互。
Dataset类通过内部包含对象的类型进行参数化，如Java的Dataset ，Scala的Dataset [T] 。举个例子，Dataset [Person]只包含Person类的对象。从Spark 2.0开始，受支持的类型遵循Java的JavaBean模式，或Scala的case类。这些类型之所以受限，是因为Spark需要可以自动分析类型T，并为Dataset中的表格数据创建适当模式。
Dataset的一个优点是，只有当你需要时才可以使用它们。下面例子中，我们将定义自己的数据类型，并通过某种map函数和filter函数来操作它。操作完成后， Spark可以将其自动重新转换为DataFrame，且可以用Spark包含的数百个函数对其进行进一步操作，这样可以很容易地降到较低级的API，以便于在必要时执行类型安全的编码，同时，也可以将其升级到更高级的SQL，进行更快速的分析。下面的例子，展示了如何用类型安全函数和类似于DataFrame的SQL表达式来快速编写业务逻辑。

当对Dataset使用collect或take时，它将会收集Dataset中合适类型的对象，而不是DataFrame的Row对象。这样可以更容易地保证类型安全，用分布式或本地方式更安全地执行操作，而无需更改代码。上图的逻辑是，取出国家不为加拿大的5条记录。

Structured Streaming

结构化流处理是用来处理数据流的高级API，Spark 2.2及之后版本可用。用户可以像在批处理模式下一样，用Spark的结构化API执行结构化流处理，结构化流处理可以减少等待时间，且允许增量处理。它能让用户快速从流式系统中提取有价值的信息，而且几乎不需要更改代码。用户可以按批处理模式进行设计，然后将其转换为流式作业，即增量处理数据，这使得流处理任务变得更简单。
下面举个例子，看如何使用Spark结构化流处理。我们用到销售数据集（/tree/master/data/retail-data），该数据集有特定日期和时间信息，我们将使用按天分组的文件，每个文件代表一天的数据。我们用另外一个进程来模拟持续产生的数据，假设这些数据是由零售商持续生成的，并由我们的结构化流式处理作业进行处理。这里先简单展示一个数据样本，以便看到数据格式。

我们首先将其看作静态数据集，而后创建一个DataFrame来进行分析，接着将该静态数据集创建一个schema。

tips：spark-shell中有很多命令都是以:开始，比如退出为:quit。:paste 可以让spark-shell进入复制模式，复制完代码后，按下ctrl + D开始执行代码。（有兴趣可以参考https://blog.csdn.net/Android_xue/article/details/100517574）

由于我们处理的是时间序列数据，因此需要提一下Spark是如何对数据进行分组和聚合的。在下面的例子中，我们将查看特定客户（主键为CustomerId）进行大笔交易的时间。我们会添加一个总费用列，并查看客户花费最多的时间。window函数包含每天的所有数据，它在我们的数据中，充当的是时间序列栏的窗口作用，这是一个用来处理日期和时间戳的有用工具，我们可以通过时间间隔来指定我们的需求，而Spark会把所有数据集合起来传递给用户。

如果是运行在本地模式，最好将shuffle分区数设置为更适合本地模式的数量，该配置指定了在shuffle后，应该创建的分区数量。默认值是200（但由于本书作者机器上没有足够的执行进程，他把分区数减少为5）。

了解了工作原理后，接着来看流处理代码，可以看到，代码改动很小，最大的变化是使用readStream而不是read，maxFilesPerTrigger选项指定我们每次读入的文件数。最后通过streamingDataFrame.isStreaming指令，可以看到，确实正在进行流处理。

我们对流数据执行与之前静态DF一样的业务逻辑（按时间窗口统计花费）。

由于是一个lazy操作，我们要调用流action操作来执行此数据流。流action与静态action有所不同，因为首先要将流数据缓存到某个地方，而不是像对静态数据那样直接调用count函数（对流数据没有任何意义）。流数据将被缓存到内存上的一个数据表里，在每次被触发器触发后更新这个内存缓存。上面将maxFilesPerTrigger设置为1，表示每次读完一个文件后都会被触发，Spark会基于新读入的文件更新内存数据表的内容，以便于聚合操作始终维护着历史数据的最大值。启动数据流后，可以查看结果是否已被写入结果的接收器。

Machine Learning and Advance Analysis

Spark使用MLlib机器学习算法内置库，来支持大规模机器学习。Mllib支持对数据进行预处理、整理、模型训练和大规模预测，甚至可以用MLlib中训练的模型在结构化流处理中对流数据进行预测。 Spark提供了一个复杂的机器学习API，用于执行各种机器学习任务，从分类到回归，从聚类到深度学习。为了说明这个功能，我们将使用称为k-means的标准聚类算法对数据执行一些基本的聚类操作。k-means是一种聚类算法。首先从数据中随机选出k个初始聚类中心，最接近某个中心的那些点被分配到一个聚类里，并根据分配到该聚类的点计算它们的中心，这个中心被称为centroid。然后，将最接近该centroid的点标记为属于该centroid的点，并根据分配到某个centroid的点群计算新的中心用来更新centroid。重复这个过程来进行有限次的迭代，或者直到收敛（中心点停止变化）。
Spark准备了许多内置的预处理方法，下面将演示这些预处理方法，这些预处理方法将原始数据转换为合适的数据格式，它将在之后用于实际训练模型中，并进一步进行预测。
之前的案例，其数据由多种不同类型表示，包括时间戳、整数和字符串等。而MLlib中的机器学习算法要求将数据表示为数值形式，因此我们需要将这些数据进行转换。下面我们将用几个DataFrame转换来处理日期数据。

我们也要将数据分成训练和测试集。在该例中，我们手动将某个购买日期前的数据作为训练集，之后的数据为测试集。我们也可以用MLlib的转换API，通过训练验证分割或交叉验证来创建训练和测试集。现在我们已经准备好了数据，再把它分成一个训练集和一个测试集。由于这是一组时间序列数据，因此我们在数据集中选择一个一个的日期作为分割，虽然这可能不是训练集和测试集的最佳分割，但对于当前的这个例子来说，这种分割已经足够好了。可以看到我们的数据集被大致分为两部分。这些转换是DataFrame转换，Spark的Mllib也有提供一些转换，可以用它们自动化一些常用转换。例如StringIndexer。这将使每周的星期几转换成相应的数值，比如将星期六表示为6，但是，通过此编号方案，我们隐式指出星期六大于星期一（因为6大于1），这显然不正确。为了解决该问题，还要使用一个OneHotEncoder来将每个值编码为其原来对应的列，这些布尔变量标识了该数值是否为与星期几相关的日子。其中每一个都会产生一组列，我们将它们组合成一个向量。 Spark中的机器学习算法输入都为vector类型，即一组数值。我们有三个关键特征：价格、数量和星期几。接下来，把这些操作设置为流水线处理模式，就可以通过完全相同的流程对未来新产生的数据进行转换。训练的准备过程要两步，首先需要为我们的数据设置合适的转换操作，且StringIndexer需要知道有多少非重复值，这样才能对应每个字符串一个数值，另外编码操作很容易，但Spark必须查看要索引的列中存在的所有不同值，这样才可以在稍后存储这些值。在配置好了训练数据后，下一步是采用流水线处理模型完成整个数据的预处理过程，以持续的和可重复的模式来转换我们的所有数据。我们可以将模型训练过程也加入到流水线处理过程中，但我们不这样做是为了缓存整个训练数据，以便于对模型训练过程中的参数进行调整，避免持续重复训练过程中的转换操作。缓存过程能将中间转换数据集的副本立即放入内存，使我们可以用较低代价反复访问数据，这远比重新运行整个流水线处理得到训练数据集节省开销。我们现在有了一套训练数据集，可以开始训练这个模型了。首先，导入我们想用的相关模型包，并对其实例化。在Spark中训练机器学习模型是具有两阶段。首先，需要初始化一个未经训练的模型，然后进行训练。在MLlib的DataFrame API中，每种算法都有两种类型，对于未经训练的算法版本，它们遵循“XX Algorithm”的命名方式，对于训练后的算法版本，我们使用“XXX AlgorithmModel”的命名方式。在我们的例子中，就是未训练的“KMeans”和训练完的“KMeansModel”。MLlib的DataFrame API中的预测器与我们之前看到的像StringIndexer这样的预处理转换操作使用大致相同的接口，它使得整个流水线处理过程（包括模型训练）变得简单。在这里，我们希望一步一步地解释，所以在这个例子中我们选择不把模型训练包含到流水线处理过程中。在训练完这个模型后，我们可以使用kmModel.computeCost(transformedTraining)，根据训练集的一些评价指标来评估开销。处理这个数据集带来的开销实际上相当高，这可能是由于我们的预处理和数据扩展部分没有做好，当然，我们可以继续改进这个模型，执行更多的预处理过程，以及执行超参数调整，确保获得一个更好的模型。

Lower-Level APIs

Spark包含很多低级原语，以支持用RDD对任意Java和Python对象进行操作，事实上， Spark所有对象都建立在RDD之上。DataFrame操作也是基于RDD的，这些高级操作会被编译到较低级的RDD上执行，以便实现极其高效的分布式执行。有些时候你可能会使用RDD，尤其是在读取或操作原始数据时，但大多时候你应该坚持用高级的结构化API。 RDD比DataFrame更低级，因为它向终端用户暴露物理执行特性（如分区）。可以用RDD来并行化已经存储在驱动器机器内存中的原始数据。例如，让我们并行化一些简单的数字并创建一个DataFrame，我们可以将RDD转换为DataFrame，以便与其他DataFrame一起使用它。

RDD可以在Scala和Python中使用，但它们并不完全等价，这与DataFrame API（执行特性相同）有所不同，这是由于RDD某些底层实现细节导致的区别。 Spark最新版本基本上没有RDD的实例，所以除了处理一些非常原始的未处理和非结构化数据之外，你应该使用结构化API而不是RDD。

SparkR

SparkR是一个在Spark上运行的R语言工具，它具有与Spark其他支持语言相同的设计准原则。要使用SparkR，只需将SparkR库导入到环境中并运行。它与Python API非常相似，只是它遵循R的语法而不是Python的。大多时候， SparkR支持Python支持的所有功能。

Spark’s Ecosystem and Packages

Spark最好的地方就在于开源社区维护的工具包和支持它的生态系统，工具在成熟并广泛使用后甚至可以直接进入Spark的核心项目。我们可以在https://spark-packages.org/找到Spark Packages的索引，所有用户都可以将自己开发的工具包发布到此代码库中，还可以在网上（如GitHub）找到各种其他项目和工具包。

Conclusion

本章展示了将Spark应用到业务与技术中的多种方法， Spark简单强大的编程模型使其可以轻松应用于各类问题的处理，并且还有大量开发人员创建围绕它的软件包，这帮助Spark可以处理大量的业务问题与挑战。随着生态系统和社区的增长，越来越多的工具包可能会持续出现。