PySpark学习笔记（1） - 好文

最近工作有需要用到PySpark，而自己对这方面还不太熟悉。有鉴于此，很有必要系统性的梳理下PySpark的基本语法和特性，以供以后参考。

1.Spark简介

Apache Spark是一个开源、强大的的分布式查询和处理引擎，最初由Matei Zaharia在UC
Berkeley读博期间开发的[1]。最早的Spark版本于2012年发布，后来被捐赠给Apache
SoftwareFoundation，成为Apache的旗舰项目之一（github链接：https://github.com/apache/spark）。

Apache Spark提供MapReduce的灵活性和可扩展性，但处理速度明显高于MapReduce，内存内处理速度和访问磁盘速度
分别比Hadoop快100倍和10倍。

Apache Spark 允许用户读取、转换和聚合数据，可以轻松地训练和部署复杂的统计模型，并且Spark
API支持Java、Scala、Python、R和SQL的访问。其可以通过如Jupyter、Apache
Zeppelin、Spark-Notebook、Databricks notebooks这样的笔记本交互式地执行快速的数据分析。除此之外，Apache
Spark还提供有几个已经实现并调优过的算法、统计模型和框架，如用于机器学习的MLlib和ML，用于图形处理的GraphX和GraphFrames，用于
处理实时流数据的Spark Streaming。

2.Spark概述

2.1 Spark 作业
任何Spark应用程序都会分离主节点上的单个驱动进程（包含若干个作业），然后将执行进程（包含若干个任务）分配给多个工作节点
。驱动进程会确定任务进程的数量和组成，然后任务进程根据DAG（有向无环图）调度器的依赖关系将任务分配给执行节点。

2.2 Spark RDD

RDD（弹性分布式数据集，Resilient Distributed Datasets）是Spark中的抽象数据结构类型
，任何数据在Spark中都被表示为RDD。RDD提供了一种高度受限的共享内存模型，即RDD是只读的记录分区的集合
，只能通过在其他RDD执行确定的转换操作（如map、join和group by）而创建，其本质是不可变Java虚拟机（JVM）对象的分布式集合。在PySpark
中，Python数据就是存储在这些JVM对象中的。

RDD主要有两组操作：转换操作（transformation,返回指向新RDD的指针）和行动操作（action,在运行计算后向驱动程序返回值）。

常见的RDD基础操作API如下表所示：

操作类型

函数名

作用

转化操作

map()

参数是函数，函数应用于RDD每一个元素，返回值是新的RDD

flatMap()

参数是函数，函数应用于RDD每一个元素，将元素数据进行拆分，变成迭代器，返回值是新的RDD

filter()

参数是函数，函数会过滤掉不符合条件的元素，返回值是新的RDD

distinct()

没有参数，将RDD里的元素进行去重操作

union()

参数是RDD，生成包含两个RDD所有元素的新RDD

intersection()

参数是RDD，求出两个RDD的共同元素

subtract()

参数是RDD，将原RDD里和参数RDD里相同的元素去掉

cartesian()

参数是RDD，求两个RDD的笛卡儿积

行动操作

collect()

返回RDD所有元素

count()

RDD里元素个数

countByValue()

各元素在RDD中出现次数

reduce()

并行整合所有RDD数据，例如求和操作

fold(0)(func)

和reduce功能一样，不过fold带有初始值

aggregate(0)(seqOp,combop)

和reduce功能一样，但是返回的RDD数据类型和原RDD不一样

foreach(func)

对RDD每个元素都是使用特定函数

2.3 DataFrame

DataFrame是一种分布在集群节点中的不可变的分布式数据集，这种数据集是以RDD为基础的，其被组织成指定的列，类似于关系数据库的二维表格和Python中的
Pandas.DataFrame，DataFrame本身带有Schema元信息，即DataFrame所表示的二维表数据集的每一列都带有名称和类型。如下图所示：

DataFrame的一个主要优点是：Spark引擎一开始就构建了一个逻辑执行计划，而且执行生成的代码是基于成本优化程序确定的物理计划，DataFrame的引入
能够显著提高PySpark中的查询速度。

2.4 SparkSession

SparkSession是Spark2.0新引入的概念，为用户提供了统一的切入点，来让用户学习Spark的各项功能。

在Spark的早期版本（Spark1.x）中，SparkContext
是Spark的主要切入点。在当时，RDD是Spark主要的API，可以直接通过SparkContext来创建和操作RDD，但对于其他的API，则需要使用不同的context。如：对于
sql，使用SQLContext；对于hive，使用hiveContext；对于Streaming，使用StreamingContext
。但是随着版本的迭代，DataFrame和DataSet的API逐渐成为标准的API，就需要为它们建立新的切入点。在Spark2.0中，引入SparkSession作为DataFrame和DataSet的API的切入点，其内部封装了
SparkConf、SparkContext和SQLContext。下面是PySpark中创建SparkSession的Demo：
from pyspark.sql import SparkSession #使用builder模式创建SparkSession spark =
SparkSession.builder \ .master("local") \ #设置为local .appName("SparkSession
Example") \ #指定应用程序的名字 .enableHiveSupport() \ #使SparkSession支持Hive
.getOrCreate() #创建或获取一个SparkSession

上面的示例代码使用builder模式来创建SparkSession。如果SparkContext已经存在，SparkSession就会重用它；如果不存在，Spark就会创建一个新的SparkContext。在每一个JVM中只能有一个SparkContext，但是在一个Spark程序中可以有多个SparkSession。

[1]Learning PySpark，Tomasz Drabas(中文名：PySpark 实战指南,译者：栾云杰)

热门工具换一换