Apache Spark 教程：机器学习

面向大数据与机器学习的 Apache Spark 与 Python

Apache Spark 以其快速、易用且通用的大数据处理引擎闻名，内置了用于流处理、SQL、机器学习（ML）和图处理的模块。这项技术是数据工程师的热门技能，同时数据科学家在进行探索性数据分析（EDA）、特征提取以及当然还有机器学习时，也能从学习 Spark 中受益。

在本教程中，您将通过 PySpark（Spark 的 Python API，将 Spark 编程模型暴露给 Python）来使用 Spark。更具体地说，您将重点了解：

在本地个人电脑上安装 PySpark，并进行设置以便使用交互式 Spark shell 对数据进行一些快速的交互式分析。您将看到如何使用 pip、Homebrew 以及通过 Spark 下载页面来完成这一步。
学习 Spark 的基础用法：您将看到如何创建 RDD，并在其上执行基本操作。
在 Jupyter Notebook 中开始使用 PySpark，并加载一个真实的数据集。
借助 DataFrame 来探索和预处理第一步中加载的数据，这需要使用 Spark SQL，它允许您在 Spark 程序中查询结构化数据。
使用 Spark ML 创建一个线性回归模型并将数据馈入，之后您将能够进行预测。最后，
评估您构建的机器学习模型。

如果您更感兴趣的是在 R 中使用 Spark，可以查看 DataCamp 的免费课程使用 sparklyr 入门 R 中的 Spark，或下载 PySpark SQL 速查表。

安装 Apache Spark

安装 Spark 并让其运行可能具有一定挑战。本节将介绍在您的电脑上完成安装的一些步骤。

首先，您需要检查是否满足前置条件。Spark 使用 Scala 编写，并运行在 Java 虚拟机（JVM）环境中。这就是为什么您需要确认是否已安装 Java Development Kit（JDK）。之所以要这样做，是因为 JDK 会为您提供一个或多个 JVM 实现。通常，您应选择最新版本，在撰写本文时是 JDK8。

接下来，就可以下载 Spark 了！

使用 `pip` 下载 `pyspark`

随后，您可以借助 pip 下载并安装 PySpark。这非常简单，和安装其他软件包类似。只需运行常用命令，繁重工作就会自动完成：

$ pip install pyspark技术支持

或者，您也可以前往 Spark 下载页面。在前三步保持默认选项，您将在第 4 步找到一个可下载的链接。点击该链接进行下载。本教程中，您将下载 2.2.0 版 Spark 发行包，以及“Pre-built for Apache Hadoop 2.7 and later”类型的预编译包。

注意：下载可能需要一些时间！

使用 Homebrew 下载 Spark

您也可以使用 Homebrew（一个免费且开源的包管理器）安装 Spark。这在使用 macOS 时尤其方便。

只需运行以下命令即可搜索 Spark、获取更多信息，并最终在个人电脑上完成安装：

# Search for spark
$ brew search spark

# Get more information on apache-spark
$ brew info apache-spark

# Install apache-spark
$ brew install apache-spark技术支持

下载并设置 Spark

接下来，请确保对出现在 Downloads 文件夹中的压缩包进行解压。您可以双击 spark-2.2.0-bin-hadoop2.7.tgz 归档文件让系统自动为您解压，或者打开终端并运行以下命令：

$ tar xvf spark-2.2.0-bin-hadoop2.7.tgz技术支持

然后，将解压后的文件夹移动到 /usr/local/spark，运行以下命令：

$ mv spark-2.1.0-bin-hadoop2.7 /usr/local/spark技术支持

注意：如果移动该文件夹到新位置时出现权限被拒绝的错误，请在命令前加上 sudo。上面的命令将变为 $ sudo mv spark-2.1.0-bin-hadoop2.7 /usr/local/spark。系统会提示您输入密码，通常就是您开机解锁电脑时使用的那个密码 :)

现在一切就绪，打开位于 /usr/local/spark 路径下的 README 文件。您可以执行

$ cd /usr/local/spark技术支持

这会将您带到所需的文件夹。接着，您可以开始查看文件夹并阅读其中包含的 README 文件。

首先，使用 $ ls 获取该 spark 文件夹中的文件和子文件夹列表。您会看到里面有一个 README.md 文件。您可以通过执行以下任一命令打开它：

# Open and edit the file
$ nano README.md

# Just read the file 
$ cat README.md技术支持

提示：在键入文件名时，可以使用键盘上的 Tab 键进行自动补全，这能为您节省一些时间。

您会发现这个 README 提供了有关 Spark 的一些常规信息、在线文档、构建 Spark、交互式 Scala 和 Python shell、示例程序等内容。

这里可能最让您感兴趣的是如何构建 Spark 的部分，但请注意，只有当您未下载预编译版本时，这才特别相关。然而在本教程中，您下载的是预编译版本。您可以按 CTRL + X 退出 README，回到 spark 文件夹。

如果您选择的是尚未构建的版本，请确保运行 README 文件中列出的命令。在撰写本文时，这是以下命令：

$ build/mvn -DskipTests clean package run技术支持

请注意，该命令可能需要一段时间才能运行完成。

PySpark 基础：RDD

既然您已经成功安装了 Spark 和 PySpark，让我们先通过探索交互式 Spark Shell 来入门，并掌握开始所需的一些基础知识。不过在本教程的其余部分，您将通过 Jupyter notebook 使用 PySpark。

Spark 应用程序与 Spark Shell 的区别

交互式 shell 是一个“读取-求值-打印-循环”（REPL）环境；这意味着无论您输入什么，系统都会读取、求值并将结果打印给您，以便您继续分析。这可能让您想到 IPython，这是一个功能强大的交互式 Python shell，您或许在使用 Jupyter 时已接触过。若想了解更多，建议阅读 DataCamp 的博文 IPython 或 Jupyter。

这意味着您可以使用适用于 Python 和 Scala 的 shell 来完成需要进行的所有交互式工作。

除了 shell，您还可以编写并部署 Spark 应用程序。与编写应用程序不同，在 shell 中 SparkSession 已为您创建，您可直接开始工作，不必在创建它上浪费宝贵时间。

现在您可能会问：什么是 SparkSession？

它是 Spark 功能的主要入口：代表与 Spark 集群的连接，您可以用它在集群上创建 RDD 并广播变量。使用 Spark 时，一切都从 SparkSession 开始并以它结束。注意：在 Spark 2.0.0 之前，三个主要的连接对象是 SparkContext、SqlContext 和 HiveContext。

稍后您会看到更多内容。现在，我们先专注于 shell。

Python Spark Shell

在位于 /usr/local/spark 的 spark 文件夹中，您可以运行

$ ./bin/pyspark技术支持

起初，您会看到一些文本输出。随后会出现 “Spark”，类似如下：

Python 2.7.13 (v2.7.13:a06454b1afa1, Dec 17 2016, 12:39:47) 
[GCC 4.2.1 (Apple Inc. build 5666) (dot 3)] on darwin
Type "help", "copyright", "credits" or "license" for more information.
Using Spark's default log4j profile: org/apache/spark/log4j-defaults.properties
Setting default log level to "WARN".
To adjust logging level use sc.setLogLevel(newLevel). For SparkR, use setLogLevel(newLevel).
17/07/26 11:41:26 WARN NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable
17/07/26 11:41:47 WARN ObjectStore: Failed to get database global_temp, returning NoSuchObjectException
Welcome to
      ____              __
     / __/__  ___ _____/ /__
    _\ \/ _ \/ _ `/ __/  '_/
   /__ / .__/\_,_/_/ /_/\_\   version 2.2.0
      /_/

Using Python version 2.7.13 (v2.7.13:a06454b1afa1, Dec 17 2016 12:39:47)
SparkSession available as 'spark'.
>>>技术支持

当您看到这些时，就说明您可以在交互式 shell 中开始实验了！

提示：如果您更喜欢使用 IPython shell 而不是 Spark shell，可以通过设置以下环境变量来实现：

export PYSPARK_DRIVER_PYTHON="/usr/local/ipython/bin/ipython"技术支持

创建 RDD

现在，从一个小例子开始，创建一个 RDD，它是 Spark 最基本的构建块。RDD 仅表示数据，但它不是单个对象、记录集合、结果集或数据集。这是因为它面向的是分布在多台计算机上的数据：同一个 RDD 可以分布在数千个 JVM 上，因为 Spark 在幕后会自动对数据进行分区以实现并行。当然，您可以调整并行度以获得更多分区。这就是为什么 RDD 实际上是分区的集合。

您可以使用 parallelize() 函数并传入一些数据（一个可迭代对象，如 list 或集合）来轻松创建一个简单的 RDD：

>>> rdd1 = spark.sparkContext.parallelize([('a',7),('a',2),('b',2)])
>>> rdd2 = spark.sparkContext.parallelize([("a",["x","y","z"]), ("b",["p", "r"])])
>>> rdd3 = spark.sparkContext.parallelize(range(100))技术支持

注意：SparkSession 对象中包含 SparkContext 对象，您可以通过 spark.sparkContext 访问。出于向后兼容的原因，也仍可使用 sc 来调用 SparkContext，例如 rdd1 = sc.parallelize(['a',7),('a',2),('b',2)])。

RDD 操作

现在您已创建了 RDD，可以并行地对 rdd1 和 rdd2 中的分布式数据进行操作。操作分为两类：转换（transformations）和行动（actions）。

为了更直观地区分二者，常见的转换包括 map()、filter()、flatMap()、sample()、randomSplit()、coalesce() 和 repartition()；常见的行动包括 reduce()、collect()、first()、take()、count()、saveAsHadoopFile()。

转换是对 RDD 的惰性操作，会创建一个或多个新的 RDD；而行动会生成非 RDD 的值：它们返回结果集、数字、文件等。

例如，您可以使用下面这个简单的 lambda 函数聚合 rdd1 的所有元素，并将结果返回给 driver 程序：

>>> rdd1.reduce(lambda a,b: a+b)技术支持

执行这行代码将得到如下结果：('a', 7, 'a', 2, 'b', 2)。另一个转换示例是 flatMapValues()，它运行在键值对 RDD（如 rdd2）上。在此情形下，您将键值对 RDD rdd2 中的每个值传入一个 flatMap 函数，而不更改键，即下面定义的 lambda 函数；之后通过 collect() 执行动作收集结果。

>>> rdd2.flatMapValues(lambda x: x).collect()
[('a', 'x'), ('a', 'y'), ('a', 'z'), ('b', 'p'), ('b', 'r')]技术支持

数据

在通过交互式 shell 学了一些基础之后，是时候开始使用一些真实数据了。在本教程中，您将使用 California Housing 数据集。当然要注意，这其实是“小”数据，在此场景下使用 Spark 可能有些大材小用；本教程仅用于教学目的，旨在向您展示如何使用 PySpark 构建机器学习模型。

加载与探索数据

即使您对数据已有一定了解，也应花些时间更深入地探索；不过在此之前，您将把 Jupyter Notebook 与 Spark 配置好，并着手定义 SparkContext。

在 Jupyter Notebook 中使用 PySpark

在本部分中，您不会使用 ishell，而是构建自己的应用。您将在 Jupyter Notebook 中进行。所需的一切您都已安装好，因此让 PySpark 在 Jupyter 中运行无需太多额外操作。

您可以像往常一样启动 notebook 应用，运行 $ jupyter notebook。然后新建一个 notebook，简单地导入 findspark 库并使用 init() 函数。在此例中，您将向 init() 传入路径 /usr/local/spark，因为您确定 Spark 安装在该路径下。

# Import findspark 
import findspark

# Initialize and provide path
findspark.init("/usr/local/spark")

# Or use this alternative
#findspark.init()技术支持

提示：如果不确定路径是否设置正确，或不清楚 Spark 安装在何处，您可以使用 findspark.find() 自动检测 Spark 的安装位置。

如果您在寻找在 Jupyter 中使用 Spark 的其他方式，请参阅我们的Apache Spark in Python: Beginner’s Guide。

现在一切就绪，终于可以开始创建您的第一个 Spark 程序了！

创建第一个 Spark 程序

首先要做的是从 pyspark 包中导入 SparkContext 并初始化它。请记住：在此之前您无需这样做，因为交互式 Spark shell 会自动为您创建并初始化！在这里，您需要自己多做一点工作 :)

从 pyspark.sql 导入 SparkSession 模块，并通过 builder() 方法构建一个 SparkSession。之后，您可以设置要连接的 master URL、应用名称，添加一些附加配置（例如 executor 内存），最后使用 getOrCreate() 获取当前正在运行的 Spark 会话，或在没有会话运行时创建一个。

# Import SparkSession
from pyspark.sql import SparkSession

# Build the SparkSession
spark = SparkSession.builder \
   .master("local") \
   .appName("Linear Regression Model") \
   .config("spark.executor.memory", "1gb") \
   .getOrCreate()
   
sc = spark.sparkContext技术支持

注意：如果您遇到类似这样的 FileNotFoundError：“No such file or directory: ‘/User/YourName/Downloads/spark-2.1.0-bin-hadoop2.7/./bin/spark-submit’”，说明需要（重新）设置 Spark PATH。通过执行 $ cd 回到主目录，然后运行 $ nano .bash_profile 编辑 .bash_profile 文件。

在文件底部添加如下内容

export SPARK_HOME="/usr/local/spark"技术支持

使用 CTRL + X 退出文件，但请确保保存更改，输入 Y 确认修改。接着，不要忘了通过运行 source .bash_profile 使更改生效。

提示：如果需要，您还可以设置其他环境变量；您可能用不上，但知道可以按需设置总是有益的。示例如下：

# Set a fixed value for the hash seed secret
export PYTHONHASHSEED=0

# Set an alternate Python executable
export PYSPARK_PYTHON=/usr/local/ipython/bin/ipython

# Augment the default search path for shared libraries
export LD_LIBRARY_PATH=/usr/local/ipython/bin/ipython

# Augment the default search path for private libraries 
export PYTHONPATH=$SPARK_HOME/python/lib/py4j-*-src.zip:$PYTHONPATH:$SPARK_HOME/python/技术支持

另请注意：现在您已初始化了一个默认的 SparkSession。不过在大多数情况下，您会希望做进一步配置。开始处理大数据时，您会发现这非常必要。若想了解更多，请查看此页面。

加载数据

本教程使用 California Housing 数据集。它出现在 1997 年的一篇论文 Sparse Spatial Autoregressions 中，由 Pace（R. Kelley）与 Ronald Barry 撰写，发表在 Statistics and Probability Letters 期刊。研究者通过 1990 年加州人口普查数据构建了该数据集。

数据包含每个普查街区组一行。街区组是美国人口普查局发布样本数据的最小地理单元（通常人口为 600 至 3000 人）。在此样本中，一个街区组平均包含 1425.5 名居住在地理位置紧凑区域的个体。您可以从此网页或阅读上文提到的论文（见这里）获取这些信息。

这些空间数据包含关于房价的 20,640 条观测与 9 个经济变量：

Longitude 指每个街区组相对于地球赤道的南北角距离；
Latitude 指每个街区组相对于地球赤道的东西角距离；
Housing median age 为属于某街区组人群的年龄中位数。注意：中位数是观测值频率分布中处于中点的数值；
Total rooms 为每个街区组内房屋的房间总数；
Total bedrooms 为每个街区组内房屋的卧室总数；
Population 为某个街区组的居民人数；
Households 指每个街区组内的住户及其成员；
Median income 用于记录属于某街区组人群的收入中位数；以及，
Median house value 为因变量，指每个街区组的房价中位数。

此外，您还会了解到，数据中已排除了所有在自变量和因变量上为零的街区组。

Median house value 是因变量，并将在您的 ML 模型中作为目标变量。

您可以在此处下载数据。找到 houses.zip 文件夹，下载并解压，以便访问数据文件夹。

接下来，您将使用 textFile() 方法将下载文件夹中的数据读取为 RDD。该方法接收文件的 URI，这里是您本机的本地路径，并将文件按行读取为集合。为方便起见，您不仅会读取 .data 文件，还会读取包含表头的 .domain 文件。这样可以帮助您再次核对变量的顺序。

# Load in the data
rdd = sc.textFile('/Users/yourName/Downloads/CaliforniaHousing/cal_housing.data')

# Load in the header
header = sc.textFile('/Users/yourName/Downloads/CaliforniaHousing/cal_housing.domain')技术支持

数据探索

仅通过查看找到数据集的网页，您已经获取了不少信息，但最好还是亲自动手，借助 Spark（此处为 Python 版）来检查数据。

这里需要理解的是，由于 Spark 的执行是“惰性”的，目前还没有真正执行任何操作。您的数据其实还未被真正读取。rdd 和 header 变量此刻更像是概念。您需要推动 Spark 为您工作，所以让我们使用 collect() 方法查看 header：

header.collect()技术支持

collect() 方法会将整个 RDD 拉回到单机，您将看到如下结果：

[u'longitude: continuous.', u'latitude: continuous.', u'housingMedianAge: continuous. ', u'totalRooms: continuous. ', u'totalBedrooms: continuous. ', u'population: continuous. ', u'households: continuous. ', u'medianIncome: continuous. ', u'medianHouseValue: continuous. ']技术支持

提示：谨慎使用 collect()！运行这行代码可能导致 driver 内存不足。因此，如果您只想打印 RDD 的少量元素，以下使用 take() 的方式更安全。总体而言，尽可能限制结果集是个好原则，就像使用 SQL 时一样。

您会得知变量的顺序与上文数据集介绍中的顺序一致，同时也得知所有列应该是连续型数值。让我们再推动 Spark 多做点工作，查看加州房价数据来确认这一点。

对 RDD 调用 take() 方法：

rdd.take(2)技术支持

执行上述代码将取出 RDD 的前两条元素。结果如您所料：由于使用 textFile() 读取文件，所有行都整体读取了。条目以英文逗号分隔，行与行之间也以逗号分隔：

[u'-122.230000,37.880000,41.000000,880.000000,129.000000,322.000000,126.000000,8.325200,452600.000000', u'-122.220000,37.860000,21.000000,7099.000000,1106.000000,2401.000000,1138.000000,8.301400,358500.000000']技术支持

这显然需要解决。您不需要拆分条目本身，但必须确保数据的每一行都是独立的元素。为此，您将使用 map() 函数，并传入一个在英文逗号处对行进行 split 的 lambda 函数。然后，像之前一样通过 take() 检查结果：

请记住：lambda 函数是在运行时创建的匿名函数。

# Split lines on commas
rdd = rdd.map(lambda line: line.split(","))

# Inspect the first 2 lines 
rdd.take(2)技术支持

您将得到如下结果：

[[u'-122.230000', u'37.880000', u'41.000000', u'880.000000', u'129.000000', u'322.000000', u'126.000000', u'8.325200', u'452600.000000'], [u'-122.220000', u'37.860000', u'21.000000', u'7099.000000', u'1106.000000', u'2401.000000', u'1138.000000', u'8.301400', u'358500.000000']]技术支持

或者，您也可以使用以下函数来检查数据：

# Inspect the first line 
rdd.first()

# Take top elements
rdd.top(2)技术支持

如果您习惯使用 Pandas 或 R 的数据框，可能会期望看到表头，但这里并没有。为了让操作更方便，您将从 RDD 过渡到 DataFrame。只要可以，建议优先使用 DataFrame。尤其在使用 Python 时，DataFrame 的性能优于 RDD。

但二者有什么区别？

当您想对非结构化数据执行底层的转换与行动操作时，可以使用 RDD。这意味着您不打算在处理过程中施加模式（schema），也不需要按名称或按列访问属性。呼应之前关于性能的说明，使用 RDD 时，您并不一定追求 DataFrame 对（半）结构化数据所能提供的性能优势。当您更倾向于使用函数式编程构造来操作数据，而不是领域特定表达式时，可以使用 RDD。

总而言之，您现在将切换到 DataFrame，以便使用更高层的表达式、执行 SQL 查询以进一步探索数据，并获得按列访问。

那就开始吧。

第一步是创建一个 SchemaRDD，或者说是带有模式的 Row 对象的 RDD。这很正常，因为与 DataFrame 类似，您最终希望得到行和列的结构。每个条目对应某一行与某一列，列还具有数据类型。

您将再次使用 map() 函数和一个 lambda 函数，将每个条目映射到 Row 的字段。为更加直观，考虑这一行：

[u'-122.230000', u'37.880000', u'41.000000', u'880.000000', u'129.000000', u'322.000000', u'126.000000', u'8.325200', u'452600.000000']技术支持

该 lambda 函数表示您将构建一个 SchemaRDD 中的行，索引 0 的元素命名为 “longitude”，依此类推。

有了这个 SchemaRDD，您就可以使用 toDF() 方法将 RDD 轻松转换为 DataFrame。

# Import the necessary modules 
from pyspark.sql import Row

# Map the RDD to a DF
df = rdd.map(lambda line: Row(longitude=line[0], 
                              latitude=line[1], 
                              housingMedianAge=line[2],
                              totalRooms=line[3],
                              totalBedRooms=line[4],
                              population=line[5], 
                              households=line[6],
                              medianIncome=line[7],
                              medianHouseValue=line[8])).toDF()技术支持

现在您已有 DataFrame df，可以使用此前用过的方法进行检查，即 first() 和 take()，还可以使用 head() 与 show()：

# Show the top 20 rows 
df.show()技术支持

您会立刻发现这与之前的 RDD 看起来大不相同：

提示：使用 df.columns 返回 DataFrame 的列名。

数据似乎已整齐地按列排好，但数据类型如何呢？Spark 在读取数据时会尝试推断模式，但这里是否成功？使用 df.dtypes 或 df.printSchema() 来进一步了解 DataFrame 中包含的数据类型。

# Print the data types of all `df` columns
# df.dtypes

# Print the schema of `df`
df.printSchema()技术支持

由于您未执行第一行代码，您只会得到如下结果：

root
 |-- households: string (nullable = true)
 |-- housingMedianAge: string (nullable = true)
 |-- latitude: string (nullable = true)
 |-- longitude: string (nullable = true)
 |-- medianHouseValue: string (nullable = true)
 |-- medianIncome: string (nullable = true)
 |-- population: string (nullable = true)
 |-- totalBedRooms: string (nullable = true)
 |-- totalRooms: string (nullable = true)技术支持

所有列仍是字符串类型……有些让人失望！

如果您想继续使用这个 DataFrame，需要纠正这种情况，为各列指定“更合适”的数据类型。这样做也有利于性能。直觉上，您可能会采用如下方案，声明将 DataFrame df 的每一列都转为 FloatType()：

from pyspark.sql.types import *

df = df.withColumn("longitude", df["longitude"].cast(FloatType())) \
   .withColumn("latitude", df["latitude"].cast(FloatType())) \
   .withColumn("housingMedianAge",df["housingMedianAge"].cast(FloatType())) \
   .withColumn("totalRooms", df["totalRooms"].cast(FloatType())) \ 
   .withColumn("totalBedRooms", df["totalBedRooms"].cast(FloatType())) \ 
   .withColumn("population", df["population"].cast(FloatType())) \ 
   .withColumn("households", df["households"].cast(FloatType())) \ 
   .withColumn("medianIncome", df["medianIncome"].cast(FloatType())) \ 
   .withColumn("medianHouseValue", df["medianHouseValue"].cast(FloatType()))技术支持

但这种重复调用既冗长又容易出错，观感也不好。为何不写一个函数，以更整洁的方式为您完成这些工作？

下面这个用户自定义函数（UDF）接收一个 DataFrame、列名列表以及希望赋予列的新数据类型。它会对每个列名进行处理，将该列 cast 为新类型，然后返回 DataFrame：

# Import all from `sql.types`
from pyspark.sql.types import *

# Write a custom function to convert the data type of DataFrame columns
def convertColumn(df, names, newType):
  for name in names: 
     df = df.withColumn(name, df[name].cast(newType))
  return df 

# Assign all column names to `columns`
columns = ['households', 'housingMedianAge', 'latitude', 'longitude', 'medianHouseValue', 'medianIncome', 'population', 'totalBedRooms', 'totalRooms']

# Conver the `df` columns to `FloatType()`
df = convertColumn(df, columns, FloatType())技术支持

这样看起来好多了！像之前一样，您可以用 printSchema() 方法快速检查 df 的数据类型。

既然这些都已就绪，就该正式开始数据探索了。您之前看到，按列访问和 SQL 查询是使用 DataFrame 的两个优势。现在深入一些。先从小处着手，只选择 df 中两列，并只查看 10 行：

df.select('population','totalBedRooms').show(10)技术支持

该查询将得到如下结果：

+----------+-------------+
|population|totalBedRooms|
+----------+-------------+
|     322.0|        129.0|
|    2401.0|       1106.0|
|     496.0|        190.0|
|     558.0|        235.0|
|     565.0|        280.0|
|     413.0|        213.0|
|    1094.0|        489.0|
|    1157.0|        687.0|
|    1206.0|        665.0|
|    1551.0|        707.0|
+----------+-------------+
only showing top 10 rows技术支持

您也可以像下面的示例一样构造更复杂的查询：

df.groupBy("housingMedianAge").count().sort("housingMedianAge",ascending=False).show()技术支持

将得到如下结果：

+----------------+-----+                                                        
|housingMedianAge|count|
+----------------+-----+
|            52.0| 1273|
|            51.0|   48|
|            50.0|  136|
|            49.0|  134|
|            48.0|  177|
|            47.0|  198|
|            46.0|  245|
|            45.0|  294|
|            44.0|  356|
|            43.0|  353|
|            42.0|  368|
|            41.0|  296|
|            40.0|  304|
|            39.0|  369|
|            38.0|  394|
|            37.0|  537|
|            36.0|  862|
|            35.0|  824|
|            34.0|  689|
|            33.0|  615|
+----------------+-----+
only showing top 20 rows技术支持

除了查询，您也可以选择对数据进行描述，获取一些汇总统计信息。这对后续一定大有帮助！

df.describe().show()技术支持

查看所有（数值型）属性的最小值与最大值。您会发现多个属性的取值范围很宽：您需要对数据集进行标准化。

数据预处理

通过上述小规模探索性数据分析收集到的信息，您已足够对数据进行预处理，以便馈入模型。

无需担心缺失值；所有为零的条目均已从数据集中排除。
您可能需要对数据进行标准化，因为最小值与最大值的范围相当大。
可能还有一些可以添加的附加属性，例如记录每个房间的卧室数，或每个住户的房间数等特征。
您的因变量数值也较大；为更方便处理，您需要稍作调整。

预处理目标值

首先从因变量 medianHouseValue 开始。为便于处理目标值，您将把房价按 100,000 为单位表达。这意味着诸如 452600.000000 的目标应变为 4.526：

# Import all from `sql.functions` 
from pyspark.sql.functions import *

# Adjust the values of `medianHouseValue`
df = df.withColumn("medianHouseValue", col("medianHouseValue")/100000)

# Show the first 2 lines of `df`
df.take(2)技术支持

查看 take() 的结果，您可以清楚地看到数值已正确调整：

[Row(households=126.0, housingMedianAge=41.0, latitude=37.880001068115234, longitude=-122.2300033569336, medianHouseValue=4.526, medianIncome=8.325200080871582, population=322.0, totalBedRooms=129.0, totalRooms=880.0), Row(households=1138.0, housingMedianAge=21.0, latitude=37.86000061035156, longitude=-122.22000122070312, medianHouseValue=3.585, medianIncome=8.301400184631348, population=2401.0, totalBedRooms=1106.0, totalRooms=7099.0)]技术支持

特征工程

在调整了 medianHouseValue 的值之后，您还可以添加上文提到的附加变量。您将向数据集中添加以下列：

Rooms per household，指每个街区组中每个住户的房间数；
Population per household，基本反映每个街区组中每个住户居住的人数；以及
Bedrooms per room，大致反映每个街区组中房间中有多少为卧室；

由于您使用的是 DataFrame，最好使用 select() 方法选择要处理的列，即 totalRooms、households 和 population。此外，您需要通过添加 col() 函数来指明您在处理列。否则，将无法进行诸如除法的逐元素运算：

# Import all from `sql.functions` if you haven't yet
from pyspark.sql.functions import *

# Divide `totalRooms` by `households`
roomsPerHousehold = df.select(col("totalRooms")/col("households"))

# Divide `population` by `households`
populationPerHousehold = df.select(col("population")/col("households"))

# Divide `totalBedRooms` by `totalRooms`
bedroomsPerRoom = df.select(col("totalBedRooms")/col("totalRooms"))

# Add the new columns to `df`
df = df.withColumn("roomsPerHousehold", col("totalRooms")/col("households")) \
   .withColumn("populationPerHousehold", col("population")/col("households")) \
   .withColumn("bedroomsPerRoom", col("totalBedRooms")/col("totalRooms"))
   
# Inspect the result
df.first()技术支持

您会看到，对于第一行，每个住户约有 6.98 个房间，该街区组的住户平均约 2.5 人，卧室占比约 0.14：

Row(households=126.0, housingMedianAge=41.0, latitude=37.880001068115234, longitude=-122.2300033569336, medianHouseValue=4.526, medianIncome=8.325200080871582, population=322.0, totalBedRooms=129.0, totalRooms=880.0, roomsPerHousehold=6.984126984126984, populationPerHousehold=2.5555555555555554, bedroomsPerRoom=0.14659090909090908)技术支持

接下来——这也是为了预先避免在标准化数据时可能遇到的问题——您还将重新排序列。由于您不一定想要标准化目标值，因此需要确保在数据集中将其隔离出来。

在此情形下，您将使用 select() 方法，并按更合适的顺序传入列名。在这里，将目标变量 medianHouseValue 放在第一列，以免受标准化影响。

另请注意：这是排除不打算纳入分析的变量的时机。本例中，让我们去掉 longitude、latitude、housingMedianAge 和 totalRooms 等变量。

# Re-order and select columns
df = df.select("medianHouseValue", 
              "totalBedRooms", 
              "population", 
              "households", 
              "medianIncome", 
              "roomsPerHousehold", 
              "populationPerHousehold", 
              "bedroomsPerRoom")技术支持

标准化

现在您已重新排好列顺序，几乎就可以对数据进行标准化了。还差一步：将特征与目标变量分离。本质上，这意味着要把 DataFrame 中的第一列从其余列中隔离。

在此，您将使用适用于 RDD 的 map() 函数来完成。您还会看到使用了 DenseVector() 函数。稠密向量是由一个双精度数组支撑的本地向量，用于表示其条目值。换句话说，它用于在 PySpark 中存储要使用的数值数组。

接下来，您将把 input_data 再次转换为 DataFrame，并通过传入一个列表作为第二个参数来重命名列。该列表包含列名 "label" 和 "features"：

# Import `DenseVector`
from pyspark.ml.linalg import DenseVector

# Define the `input_data` 
input_data = df.rdd.map(lambda x: (x[0], DenseVector(x[1:])))

# Replace `df` with the new DataFrame
df = spark.createDataFrame(input_data, ["label", "features"])技术支持

接下来，您终于可以对数据做缩放了。您可以使用 Spark ML 来完成：该库使得在大数据上进行机器学习可扩展且易于使用。您会找到诸如 ML 算法和构建实际 ML 流水线所需的一切工具。本例中，您无需做太多预处理，因此流水线可能有些大材小用，但如果您有兴趣，务必查看此页面。

输入列为特征，缩放后的输出列将在 scaled_df 中命名为 "features_scaled"：

# Import `StandardScaler` 
from pyspark.ml.feature import StandardScaler

# Initialize the `standardScaler`
standardScaler = StandardScaler(inputCol="features", outputCol="features_scaled")

# Fit the DataFrame to the scaler
scaler = standardScaler.fit(df)

# Transform the data in `df` with the scaler
scaled_df = scaler.transform(df)

# Inspect the result
scaled_df.take(2)技术支持

来看一下 DataFrame 与结果。您会看到第三列 features_scaled 确实已添加到 DataFrame 中，您可以将其与 features 进行比较：

[Row(label=4.526, features=DenseVector([129.0, 322.0, 126.0, 8.3252, 6.9841, 2.5556, 0.1466]), features_scaled=DenseVector([0.3062, 0.2843, 0.3296, 4.3821, 2.8228, 0.2461, 2.5264])), Row(label=3.585, features=DenseVector([1106.0, 2401.0, 1138.0, 8.3014, 6.2381, 2.1098, 0.1558]), features_scaled=DenseVector([2.6255, 2.1202, 2.9765, 4.3696, 2.5213, 0.2031, 2.6851]))]技术支持

注意：这些代码与在 Scikit-Learn 中的做法非常相似。

使用 Spark ML 构建机器学习模型

预处理完成后，终于可以开始构建线性回归模型了！与往常一样，先将数据拆分为训练集和测试集。好在使用 randomSplit() 方法即可轻松完成：

# Split the data into train and test sets
train_data, test_data = scaled_df.randomSplit([.8,.2],seed=1234)技术支持

您传入一个包含两个数字的列表，表示训练集与测试集的大小比例，并指定一个随机种子，出于可复现性的考虑。若想了解更多，可参考 DataCamp 的 Python 机器学习教程。

然后，直接开始构建模型吧！

注意：参数 elasticNetParam 对应 α（竖截距），而 regParam（正则化参数）对应 λ。详情请参见这里。

# Import `LinearRegression`
from pyspark.ml.regression import LinearRegression

# Initialize `lr`
lr = LinearRegression(labelCol="label", maxIter=10, regParam=0.3, elasticNetParam=0.8)

# Fit the data to the model
linearModel = lr.fit(train_data)技术支持

有了模型，您就可以为测试数据生成预测：使用 transform() 方法对 test_data 的标签进行预测。随后，您可以使用 RDD 操作从 DataFrame 中提取预测值与真实标签，并将二者 zip 起来得到名为 predictionAndLabel 的列表。

最后，您可以通过方括号 [] 简单访问该列表，查看预测值与真实值：

# Generate predictions
predicted = linearModel.transform(test_data)

# Extract the predictions and the "known" correct labels
predictions = predicted.select("prediction").rdd.map(lambda x: x[0])
labels = predicted.select("label").rdd.map(lambda x: x[0])

# Zip `predictions` and `labels` into a list
predictionAndLabel = predictions.zip(labels).collect()

# Print out first 5 instances of `predictionAndLabel` 
predictionAndLabel[:5]技术支持

您将看到如下真实值与预测值（按此顺序）：

[(1.4491508524918457, 0.14999), (1.5705029404692372, 0.14999), (2.148727956912464, 0.14999), (1.5831547768979277, 0.344), (1.5182107797955968, 0.398)]技术支持

评估模型

查看预测值是一回事，但更好的是查看一些指标，以更好地了解模型的实际表现。您可以先打印模型的系数与截距：

# Coefficients for the model
linearModel.coefficients

# Intercept for the model
linearModel.intercept技术支持

将得到如下结果：

# The coefficients
[0.0,0.0,0.0,0.276239709215,0.0,0.0,0.0]

# The intercept
0.990399577462技术支持

接着，您还可以使用 summary 属性获取 rootMeanSquaredError 和 r2：

# Get the RMSE
linearModel.summary.rootMeanSquaredError

# Get the R2
linearModel.summary.r2技术支持

RMSE（均方根误差）衡量两个数据集（预测值与观测/真实值）之间的误差。RMSE 越小，预测值与观测值越接近。
R2（“R 平方”）或决定系数衡量数据与拟合回归线的接近程度。该分数始终在 0 到 100%（此处为 0 到 1）之间，其中 0% 表示模型无法解释响应数据围绕其均值的任何变异性，而 100% 则相反：解释了所有变异性。一般而言，R² 越高，模型对数据的拟合越好。

您将得到如下结果：

# RMSE
0.8692118678997669

# R2
0.4240895287218379
技术支持

您的模型肯定还有提升空间！如果您想继续优化，可以调整传入模型的参数、原始 DataFrame 中包含的变量等……不过本教程先到这里！

在结束之前……

结束前，请确保使用以下代码停止 SparkSession：

spark.stop()技术支持

更进一步玩转大数据

恭喜！您已完成本教程，学会了如何借助 Spark ML 构建线性回归模型。

如果您有兴趣进一步学习 PySpark，建议参加 DataCamp 的 Introduction to PySpark 课程，并查看 Apache Spark 教程：使用 PySpark 进行机器学习。

主题

Python 数据科学机器学习

主题

技术

Apache Spark 教程：使用 PySpark 进行机器学习

使用 AI 探索

面向大数据与机器学习的 Apache Spark 与 Python

安装 Apache Spark

使用 `pip` 下载 `pyspark`

使用 Homebrew 下载 Spark

下载并设置 Spark

PySpark 基础：RDD

Spark 应用程序与 Spark Shell 的区别

Python Spark Shell

创建 RDD

RDD 操作

数据

加载与探索数据

在 Jupyter Notebook 中使用 PySpark

创建第一个 Spark 程序

加载数据

数据探索

数据预处理

预处理目标值

特征工程

标准化

使用 Spark ML 构建机器学习模型

评估模型

在结束之前……

更进一步玩转大数据

主题

进一步了解 Python 与 PySpark

PySpark 基础

使用 PySpark 进行机器学习

使用 PySpark 进行数据清洗

主题

技术

进一步了解 Python 与 PySpark

PySpark 基础

使用 PySpark 进行机器学习

使用 PySpark 进行数据清洗

面向大数据与机器学习的 Apache Spark 与 Python

安装 Apache Spark

使用 pip 下载 pyspark

使用 Homebrew 下载 Spark

下载并设置 Spark

PySpark 基础：RDD

Spark 应用程序与 Spark Shell 的区别

Python Spark Shell

创建 RDD

RDD 操作

数据

加载与探索数据

在 Jupyter Notebook 中使用 PySpark

创建第一个 Spark 程序

加载数据

数据探索

数据预处理

预处理目标值

特征工程

标准化

使用 Spark ML 构建机器学习模型

评估模型

在结束之前……

更进一步玩转大数据

主题

进一步了解 Python 与 PySpark

PySpark 基础

使用 PySpark 进行机器学习

使用 PySpark 进行数据清洗

进一步了解 Python 与 PySpark

PySpark 基础

使用 PySpark 进行机器学习

使用 PySpark 进行数据清洗

使用 `pip` 下载 `pyspark`