Spark调研笔记第5篇 - Spark API简介

Posted 2022-12-07 slvher

由于Spark是用Scala实现的，所以Spark天生支持Scala API，此外，还支持Java和Python API。以Spark 1.3版本的Python API为例，其模块层级关系如下图所示：

从上图可知，pyspark是Python API的顶层package，它包含了几个重要的subpackages，其中：

1) pyspark.SparkContext
它抽象了指向spark集群的一条连接，可用来创建RDD对象，它是API的主入口。

2) pyspark.SparkConf
通过它可在提交的应用代码中动态创建spark app的配置并作为conf参数传给pyspark.SparkContext实例的构造函数。若未动态创建conf，则pyspark.SparkContext实例从conf/spark-defaults.conf读取默认的全局配置。

3) pyspark.RDD
RDDs can be stored in memory between queries without requiring replication. Instead, they rebuild lost data on failure using lineage: each RDD remembers how it was built from other datasets (by transformations like map, join or groupBy) to rebuild itself.
RDD是Spark编程的核心抽象概念，它代表了一个抽象的弹性分布式数据集，Spark支持对RDD进行两类操作：transformations和actions，它们所包含的函数列表可以参考官方文档的"Transformations"和"Actions"部分。
根据Spark Programming Guide文档"RDD Operations"部分的说明，根据已经存在的数据集创建新数据集的操作被称作transformation；对数据集做计算并将结果返回driver program的操作被称作action。
例如，map是根据传入的函数参数对已有RDD做处理，其运行结果得到一个新的RDD，所以它是一个transformation操作；而reduce则是根据传入的函数参数对已有RDD做计算，计算结果不再是个RDD，而是个具体的值（对reduce来说，计算结果是个具体的数字，而其它action(s)得到的可能是个list或其他数据结构），所以reduce是个action操作。
需要特别强调的是，Spark对所有的transformations操作都采用lazy evaluation的策略，也即spark在调度时并不是对遇到的每个transformation都立即求值以得到新的RDD，而是将针对某个RDD的一系列transformations操作记录下来，只有最终遇到action操作时，Spark才会计算先前记录的每个transformations。
这种lazy evaluation的设计思路使得Spark得以更高效运行，因为调度器可以对从初始RDD到最终action操作路径上的transformations做合并或其它变换，且只有最终的action操作结果才会返回给driver program，节省了transformations操作的中间结果在集群worker node和driver program间的传输开销。
默认情况下，调用action操作时，初始RDD经过的每个transformation操作均会被执行一次，在多个actions会经过一系列相同的transformations操作时，这种recompute显得并不高效。因此，在实际开发Spark计算任务脚本时，会被多个actions共用的transformations结果最好调用persist或cache缓存起来，这样会节省不少计算时间。

4) pyspark.Broadcast
通过Broadcast广播的变量的作用域对应用所申请的每个节点上的executor进程都是可见的，而且广播后，变量会一直存在于每个worker节点的executor进程中，直到任务结束，这样可以避免RDD数据集在driver和worker节点的executor进程间频繁传输带来的开销。
尤其是对于某些用到只读共享变量的应用（如需要加载字典且所有计算节点均需访问该字典），广播可以高效地实现变量共享的目的。

5) pyspark.Accumulator
它是Spark支持的另一种变量共享的方式（第1种方式是上面介绍的Broadcast），worker节点上的进程可以通过add()操作更新变量，更新后的变量会自动传播回driver program。

6) pyspark.SparkFiles
当应用通过SparkContext.addFile()向集群提交任务用到的文件时，调用SparkFiles类的相关方法可以解析这些文件路径并访问文件。

7) pyspark.StorageLevel
它可以指定RDD的存储级别，如只使用内存、只使用磁盘、内存为主磁盘为辅，等等。详细的控制标识可以参考这里的文档。

【参考资料】
1. Spark Programming Guide - RDD Operations
2. pyspark package
3. Spark Programming Guide: RDD Transformations
4. Spark Programming Guide: RDD Actions
5. pyspark package: pyspark.StorageLevel

======================= EOF ====================