标签：数据分析 指南 val PySpark RDD 使用 Spark 数据 我们

PySpark 大数据分析实用指南（全）

原文：zh.annas-archive.org/md5/62C4D847CB664AD1379DE037B94D0AE5

译者：飞龙

协议：CC BY-NC-SA 4.0

前言

Apache Spark 是一个开源的并行处理框架，已经存在了相当长的时间。Apache Spark 的许多用途之一是在集群计算机上进行数据分析应用程序。

本书将帮助您实施一些实用和经过验证的技术，以改进 Apache Spark 中的编程和管理方面。您不仅将学习如何使用 Spark 和 Python API 来创建高性能的大数据分析，还将发现测试、保护和并行化 Spark 作业的技术。

本书涵盖了 PySpark 的安装和设置、RDD 操作、大数据清理和整理，以及将数据聚合和总结为有用报告。您将学习如何从所有流行的数据托管平台（包括 HDFS、Hive、JSON 和 S3）获取数据，并使用 PySpark 处理大型数据集，获得实际的大数据经验。本书还将帮助您在本地机器上开发原型，然后逐步处理生产环境和大规模的混乱数据。

本书的受众

本书适用于开发人员、数据科学家、业务分析师或任何需要可靠地分析大量大规模真实世界数据的人。无论您是负责创建公司的商业智能功能，还是为机器学习模型创建出色的数据平台，或者希望使用代码放大业务影响，本书都适合您。

本书涵盖的内容

第一章《安装 Pyspark 并设置开发环境》涵盖了 PySpark 的安装，以及学习 Spark 的核心概念，包括弹性分布式数据集（RDDs）、SparkContext 和 Spark 工具，如 SparkConf 和 SparkShell。

第二章《使用 RDD 将大数据导入 Spark 环境》解释了如何使用 RDD 将大数据导入 Spark 环境，使用各种工具与修改数据进行交互，以便提取有用的见解。

第三章《使用 Spark 笔记本进行大数据清理和整理》介绍了如何在笔记本应用程序中使用 Spark，从而促进 RDD 的有效使用。

第四章《将数据聚合和总结为有用报告》描述了如何使用 map 和 reduce 函数计算平均值，执行更快的平均值计算，并使用键/值对数据点的数据透视表。

第五章《使用 MLlib 进行强大的探索性数据分析》探讨了 Spark 执行回归任务的能力，包括线性回归和 SVM 等模型。

第六章《使用 SparkSQL 为大数据添加结构》解释了如何使用 Spark SQL 模式操作数据框，并使用 Spark DSL 构建结构化数据操作的查询。

第七章《转换和操作》介绍了 Spark 转换以推迟计算，然后考虑应避免的转换。我们还将使用reduce和reduceByKey方法对数据集进行计算。

第八章《不可变设计》解释了如何使用 DataFrame 操作进行转换，以讨论高度并发环境中的不可变性。

第九章《避免洗牌和减少运营成本》涵盖了洗牌和应该使用的 Spark API 操作。然后我们将测试在 Apache Spark 中引起洗牌的操作，以了解应避免哪些操作。

第十章《以正确格式保存数据》解释了如何以正确格式保存数据，以及如何使用 Spark 的标准 API 将数据保存为纯文本。

第十一章《使用 Spark 键/值 API》，讨论了可用于键/值对的转换。我们将研究键/值对的操作，并查看键/值数据上可用的分区器。

第十二章《测试 Apache Spark 作业》更详细地讨论了在不同版本的 Spark 中测试 Apache Spark 作业。

第十三章，利用 Spark GraphX API，介绍了如何利用 Spark GraphX API。我们将对 Edge API 和 Vertex API 进行实验。

充分利用本书

本书需要一些 PySpark、Python、Java 和 Scala 的基本编程经验。

下载示例代码文件

您可以从您在www.packt.com的帐户中下载本书的示例代码文件。如果您在其他地方购买了这本书，您可以访问www.packt.com/support并注册，以便文件直接通过电子邮件发送给您。

您可以按照以下步骤下载代码文件：

1. 在www.packt.com登录或注册。

2. 选择“支持”选项卡。

3. 单击“代码下载和勘误”。

4. 在搜索框中输入书名并按照屏幕上的说明操作。

下载文件后，请确保使用最新版本的解压缩或提取文件夹：

• WinRAR/7-Zip for Windows

• Zipeg/iZip/UnRarX for Mac

• 7-Zip/PeaZip for Linux

本书的代码包也托管在 GitHub 上，网址为github.com/PacktPublishing/Hands-On-Big-Data-Analytics-with-PySpark。如果代码有更新，将在现有的 GitHub 存储库上进行更新。

我们还有其他代码包，来自我们丰富的书籍和视频目录，可在github.com/PacktPublishing/上找到。请查看！

下载彩色图像

我们还提供了一个 PDF 文件，其中包含本书中使用的屏幕截图/图表的彩色图像。您可以在这里下载：www.packtpub.com/sites/default/files/downloads/9781838644130_ColorImages.pdf。

使用的约定

本书中使用了许多文本约定。

CodeInText：指示文本中的代码词、数据库表名、文件夹名、文件名、文件扩展名、路径名、虚拟 URL、用户输入和 Twitter 句柄。以下是一个例子：“将下载的WebStorm-10*.dmg磁盘映像文件挂载为系统中的另一个磁盘。”

代码块设置如下：

test("Should use immutable DF API") {
    import spark.sqlContext.implicits._
    //given
    val userData =
        spark.sparkContext.makeRDD(List(
            UserData("a", "1"),
            UserData("b", "2"),
            UserData("d", "200")
        )).toDF()

当我们希望引起您对代码块的特定部分的注意时，相关行或项目将以粗体显示：

class ImmutableRDD extends FunSuite {
    val spark: SparkContext = SparkSession
        .builder().master("local[2]").getOrCreate().sparkContext

test("RDD should be immutable") {
    //given
    val data = spark.makeRDD(0 to 5)

任何命令行输入或输出都以以下方式编写：

total_duration/(normal_data.count())

粗体：表示一个新术语、一个重要词或屏幕上看到的词。例如，菜单或对话框中的词会以这种方式出现在文本中。以下是一个例子：“从管理面板中选择系统信息。”

警告或重要说明会出现在这样的地方。

提示和技巧会出现在这样的地方。

第一章：安装 Pyspark 并设置开发环境

在本章中，我们将介绍 Spark 并学习核心概念，如 SparkContext，以及 Spark 工具，如 SparkConf 和 Spark shell。唯一的先决条件是对基本 Python 概念的了解，并且希望从大数据中寻求洞察力。我们将学习如何使用 Spark SQL 分析和发现模式，以改进我们的业务智能。此外，您将能够通过设置 PySpark 来快速迭代解决方案。在本书结束时，您将能够使用 PySpark 处理真实的混乱数据集，从而获得实际的大数据经验。

在本章中，我们将涵盖以下主题：

• PySpark 概述

• 在 Windows 上设置 Spark 和 PySpark

• Spark 和 PySpark 中的核心概念

PySpark 概述

在开始安装 PySpark 之前，PySpark 是 Spark 的 Python 接口，让我们先了解一些 Spark 和 PySpark 的核心概念。Spark 是 Apache 的最新大数据工具，可以通过简单地转到spark.apache.org/找到。它是用于大规模数据处理的统一分析引擎。这意味着，如果您有大量数据，您可以将这些数据输入 Spark 以快速创建一些分析。如果我们比较 Hadoop 和 Spark 的运行时间，Spark 比 Hadoop 快一百倍以上。它非常易于使用，因为有非常好的 API 可用于与 Spark 一起使用。

Spark 平台的四个主要组件如下：

• Spark SQL：Spark 的清理语言

• Spark Streaming：允许您提供实时流数据

• MLlib（机器学习）：Spark 的机器学习库

• GraphX（图形）：Spark 的图形库

Spark 中的核心概念是 RDD，它类似于 pandas DataFrame，或 Python 字典或列表。这是 Spark 用来在基础设施上存储大量数据的一种方式。RDD 与存储在本地内存中的内容（如 pandas DataFrame）的关键区别在于，RDD 分布在许多机器上，但看起来像一个统一的数据集。这意味着，如果您有大量数据要并行操作，您可以将其放入 RDD 中，Spark 将为您处理并行化和数据的集群。

Spark 有三种不同的接口，如下所示：

• Scala

• Java

• Python

Python 类似于 PySpark 集成，我们将很快介绍。现在，我们将从 PySpark 包中导入一些库，以帮助我们使用 Spark。我们理解 Spark 的最佳方式是查看示例，如下面的屏幕截图所示：

lines = sc.textFile("data.txt")
lineLengths = lines.map(lambda s: len(s))
totalLength = lineLengths.reduce(lambda a, b: a + b)

在上面的代码中，我们通过调用SC.textFile("data.txt")创建了一个名为lines的新变量。sc是代表我们的 Spark 集群的 Python 对象。Spark 集群是一系列存储我们的 Spark 进程的实例或云计算机。通过调用textFile构造函数并输入data.text，我们可能已经输入了一个大型文本文件，并仅使用这一行创建了一个 RDD。换句话说，我们在这里要做的是将一个大型文本文件输入到分布式集群和 Spark 中，而 Spark 会为我们处理这个集群。

在第二行和第三行，我们有一个 MapReduce 函数。在第二行，我们使用lambda函数将长度函数映射到data.text的每一行。在第三行，我们调用了一个减少函数，将所有lineLengths相加，以产生文档的总长度。虽然 Python 的lines是一个包含data.text中所有行的变量，但在幕后，Spark 实际上正在处理data.text的片段在 Spark 集群上的两个不同实例上的分布，并处理所有这些实例上的 MapReduce 计算。

Spark SQL

Spark SQL 是 Spark 平台上的四个组件之一，正如我们在本章中之前看到的。它可以用于执行 SQL 查询或从任何现有的 Hive 绝缘中读取数据，其中 Hive 也是来自 Apache 的数据库实现。Spark SQL 看起来非常类似于 MySQL 或 Postgres。以下代码片段是一个很好的例子：

#Register the DataFrame as a SQL temporary view
df.CreateOrReplaceTempView("people")

sqlDF = spark.sql("SELECT * FROM people")
sqlDF.show()

#+----+-------+
#| age|   name|
#+----+-------+
#+null|Jackson|
#|  30| Martin|
#|  19| Melvin|
#+----|-------|

您需要从某个表中选择所有列，例如people，并使用 Spark 对象，您将输入一个非常标准的 SQL 语句，这将显示一个 SQL 结果，就像您从正常的 SQL 实现中所期望的那样。

现在让我们看看数据集和数据框。数据集是分布式数据集合。它是在 Spark 1.6 中添加的一个接口，提供了 RDD 的优势。另一方面，数据框对于那些使用过 pandas 或 R 的人来说非常熟悉。数据框只是一个组织成命名列的数据集，类似于关系数据库或 Python 中的数据框。数据集和数据框之间的主要区别在于数据框有列名。可以想象，这对于机器学习工作和输入到诸如 scikit-learn 之类的东西非常方便。

让我们看看如何使用数据框。以下代码片段是数据框的一个快速示例：

# spark is an existing SparkSession
df = spark.read.json("examples/src/main/resources/people.json")
# Displays the content of the DataFrame to stdout
df.show()

#+----+-------+
#| age|   name|
#+----+-------+
#+null|Jackson|
#|  30| Martin|
#|  19| Melvin|
#+----|-------|

与 pandas 或 R 一样，read.json允许我们从 JSON 文件中输入一些数据，而df.show以类似于 pandas 的方式显示数据框的内容。

正如我们所知，MLlib 用于使机器学习变得可扩展和简单。MLlib 允许您执行常见的机器学习任务，例如特征化；创建管道；保存和加载算法、模型和管道；以及一些实用程序，例如线性代数、统计和数据处理。另一件事需要注意的是，Spark 和 RDD 几乎是不可分割的概念。如果您对 Spark 的主要用例是机器学习，Spark 现在实际上鼓励您使用基于数据框的 MLlib API，这对我们来说非常有益，因为我们已经熟悉 pandas，这意味着平稳过渡到 Spark。

在下一节中，我们将看到如何在 Windows 上设置 Spark，并设置 PySpark 作为接口。

在 Windows 上设置 Spark 和 PySpark

完成以下步骤，在 Windows 计算机上安装 PySpark：

1. 从github.com/bmatzelle/gow/releases/download/v0.8.0/Gow-0.8.0.exe下载Gnu on Windows（GOW）。

2. GOW 允许在 Windows 上使用 Linux 命令。我们可以使用以下命令来查看通过安装 GOW 允许的基本 Linux 命令：

gow --list 

这会产生以下输出：

1. 下载并安装 Anaconda。如果需要帮助，可以参考以下教程：medium.com/@GalarnykMichael/install-python-on-windows-anaconda-c63c7c3d1444。

2. 关闭先前的命令行，打开一个新的命令行。

3. 转到 Apache Spark 网站（spark.apache.org/）。

4. 要下载 Spark，请从下拉菜单中选择以下内容：

• 最近的 Spark 版本

• 适当的软件包类型

以下屏幕截图显示了 Apache Spark 的下载页面：

1. 然后，下载 Spark。下载完成后，将文件移动到您想要解压缩的文件夹中。

2. 您可以手动解压缩，也可以使用以下命令：

gzip -d spark-2.1.0-bin-hadoop2.7.tgz tar xvf spark-2.1.0-bin-hadoop2.7.tar

1. 现在，使用以下命令将winutils.exe下载到您的spark-2.1.0-bin-hadoop2.7\bin文件夹中：

curl -k -L -o winutils.exe https://github.com/steveloughran/winutils/blob/master/hadoop-2.6.0/bin/winutils.exe?raw=true

1. 确保您的计算机上已安装 Java。您可以使用以下命令查看 Java 版本：

java --version

这会产生以下输出：

1. 使用以下命令检查 Python 版本：

python --version 

这会产生以下输出：

1. 让我们编辑我们的环境变量，这样我们可以在任何目录中打开 Spark，如下所示：

setx SPARK_HOME C:\opt\spark\spark-2.1.0-bin-hadoop2.7
setx HADOOP_HOME C:\opt\spark\spark-2.1.0-bin-hadoop2.7
setx PYSPARK_DRIVER_PYTHON ipython
setx PYSPARK_DRIVER_PYTHON_OPTS notebook

将C:\opt\spark\spark-2.1.0-bin-hadoop2.7\bin添加到你的路径中。

1. 关闭终端，打开一个新的终端，并输入以下命令：

--master local[2]

PYSPARK_DRIVER_PYTHON和PYSPARK_DRIVER_PYTHON_OPTS参数用于在 Jupyter Notebook 中启动 PySpark shell。--master参数用于设置主节点地址。

1. 接下来要做的是在bin文件夹中运行 PySpark 命令：

.\bin\pyspark

这将产生以下输出：

Spark 和 PySpark 中的核心概念

现在让我们来看看 Spark 和 PySpark 中的以下核心概念：

• SparkContext

• SparkConf

• Spark shell

SparkContext

SparkContext 是 Spark 中的一个对象或概念。它是一个大数据分析引擎，允许你以编程方式利用 Spark 的强大功能。

当你有大量数据无法放入本地机器或笔记本电脑时，Spark 的强大之处就显现出来了，因此你需要两台或更多计算机来处理它。在处理数据的同时，你还需要保持处理速度。我们不仅希望数据在几台计算机上进行计算，还希望计算是并行的。最后，你希望这个计算看起来像是一个单一的计算。

让我们考虑一个例子，我们有一个包含 5000 万个名字的大型联系人数据库，我们可能想从每个联系人中提取第一个名字。显然，如果每个名字都嵌入在一个更大的联系人对象中，将 5000 万个名字放入本地内存中是困难的。这就是 Spark 发挥作用的地方。Spark 允许你给它一个大数据文件，并将帮助处理和上传这个数据文件，同时为你处理在这个数据上进行的所有操作。这种能力由 Spark 的集群管理器管理，如下图所示：

集群管理器管理多个工作节点；可能有 2 个、3 个，甚至 100 个。关键是 Spark 的技术有助于管理这个工作节点集群，你需要一种方法来控制集群的行为，并在工作节点之间传递数据。

SparkContext 让你可以像使用 Python 对象一样使用 Spark 集群管理器的功能。因此，有了SparkContext，你可以传递作业和资源，安排任务，并完成从SparkContext到Spark 集群管理器的下游任务，然后Spark 集群管理器完成计算后将结果带回来。

让我们看看这在实践中是什么样子，以及如何设置 SparkContext：

1. 首先，我们需要导入SparkContext。

2. 创建一个新对象，将其赋给变量sc，代表使用SparkContext构造函数的 SparkContext。

3. 在SparkContext构造函数中，传递一个local上下文。在这种情况下，我们正在研究PySpark的实际操作，如下所示：

from pyspark import SparkContext
sc = SparkContext('local', 'hands on PySpark')

1. 一旦我们建立了这一点，我们只需要使用sc作为我们 Spark 操作的入口点，就像下面的代码片段中所演示的那样：

visitors = [10, 3, 35, 25, 41, 9, 29] df_visitors = sc.parallelize(visitors) df_visitors_yearly = df_visitors.map(lambda x: x*365).collect() print(df_visitors_yearly)

让我们举个例子；如果我们要分析我们服装店的虚拟数据集的访客数量，我们可能有一个表示每天访客数量的visitors列表。然后，我们可以创建一个 DataFrame 的并行版本，调用sc.parallelize(visitors)，并输入visitors数据集。df_visitors然后为我们创建了一个访客的 DataFrame。然后，我们可以映射一个函数；例如，通过映射一个lambda函数，将每日数字（x）乘以365，即一年中的天数，将其推断为一年的数字。然后，我们调用collect()函数，以确保 Spark 执行这个lambda调用。最后，我们打印出df_visitors_yearly。现在，我们让 Spark 在幕后处理我们的虚拟数据的计算，而这只是一个 Python 操作。

Spark shell

我们将返回到我们的 Spark 文件夹，即spark-2.3.2-bin-hadoop2.7，然后通过输入.\bin\pyspark来启动我们的 PySpark 二进制文件。

我们可以看到我们已经在以下截图中启动了一个带有 Spark 的 shell 会话：

现在，Spark 对我们来说是一个spark变量。让我们在 Spark 中尝试一件简单的事情。首先要做的是加载一个随机文件。在每个 Spark 安装中，都有一个README.md的 markdown 文件，所以让我们将其加载到内存中，如下所示：

text_file = spark.read.text("README.md")

如果我们使用spark.read.text然后输入README.md，我们会得到一些警告，但目前我们不必太担心这些，因为我们将在稍后看到如何解决这些问题。这里的主要问题是我们可以使用 Python 语法来访问 Spark。

我们在这里所做的是将README.md作为spark读取的文本数据放入 Spark 中，然后我们可以使用text_file.count()来让 Spark 计算我们的文本文件中有多少个字符，如下所示：

text_file.count()

从中，我们得到以下输出：

103

我们还可以通过以下方式查看第一行是什么：

text_file.first()

我们将得到以下输出：

Row(value='# Apache Spark')

现在，我们可以通过以下方式计算包含单词Spark的行数：

lines_with_spark = text_file.filter(text_file.value.contains("Spark"))

在这里，我们使用filter()函数过滤了行，并在filter()函数内部指定了text_file_value.contains包含单词"Spark"，然后将这些结果放入了lines_with_spark变量中。

我们可以修改上述命令，简单地添加.count()，如下所示：

text_file.filter(text_file.value.contains("Spark")).count()

现在我们将得到以下输出：

20

我们可以看到文本文件中有20行包含单词Spark。这只是一个简单的例子，展示了我们如何使用 Spark shell。

SparkConf

SparkConf 允许我们配置 Spark 应用程序。它将各种 Spark 参数设置为键值对，通常会使用SparkConf()构造函数创建一个SparkConf对象，然后从spark.*底层 Java 系统中加载值。

有一些有用的函数；例如，我们可以使用sets()函数来设置配置属性。我们可以使用setMaster()函数来设置要连接的主 URL。我们可以使用setAppName()函数来设置应用程序名称，并使用setSparkHome()来设置 Spark 将安装在工作节点上的路径。

您可以在spark.apache.org/docs/0.9.0/api/pyspark/pysaprk.conf.SparkConf-class.html了解更多关于 SparkConf 的信息。

摘要

在本章中，我们学习了 Spark 和 PySpark 中的核心概念。我们学习了在 Windows 上设置 Spark 和使用 PySpark。我们还介绍了 Spark 的三大支柱，即 SparkContext、Spark shell 和 SparkConf。

在下一章中，我们将学习如何使用 RDD 将大数据导入 Spark 环境。

第二章：使用 RDD 将大数据导入 Spark 环境

主要是，本章将简要介绍如何使用弹性分布式数据集（RDDs）将大数据导入 Spark 环境。我们将使用各种工具来与和修改这些数据，以便提取有用的见解。我们将首先将数据加载到 Spark RDD 中，然后使用 Spark RDD 进行并行化。

在本章中，我们将涵盖以下主题：

• 将数据加载到 Spark RDD 中

• 使用 Spark RDD 进行并行化

• RDD 操作的基础知识

将数据加载到 Spark RDD 中

在本节中，我们将看看如何将数据加载到 Spark RDD 中，并将涵盖以下主题：

• UCI 机器学习数据库

• 从存储库将数据导入 Python

• 将数据导入 Spark

让我们首先概述一下 UCI 机器学习数据库。

UCI 机器学习库

我们可以通过导航到archive.ics.uci.edu/ml/来访问 UCI 机器学习库。那么，UCI 机器学习库是什么？UCI 代表加州大学尔湾分校机器学习库，它是一个非常有用的资源，可以获取用于机器学习的开源和免费数据集。尽管 PySpark 的主要问题或解决方案与机器学习无关，但我们可以利用这个机会获取帮助我们测试 PySpark 功能的大型数据集。

让我们来看一下 KDD Cup 1999 数据集，我们将下载，然后将整个数据集加载到 PySpark 中。

将数据从存储库加载到 Spark

我们可以按照以下步骤下载数据集并将其加载到 PySpark 中：

1. 点击数据文件夹。

2. 您将被重定向到一个包含各种文件的文件夹，如下所示：

您可以看到有 kddcup.data.gz，还有 kddcup.data_10_percent.gz 中的 10%数据。我们将使用食品数据集。要使用食品数据集，右键单击 kddcup.data.gz，选择复制链接地址，然后返回到 PySpark 控制台并导入数据。

让我们看看如何使用以下步骤：

1. 启动 PySpark 后，我们需要做的第一件事是导入urllib，这是一个允许我们与互联网上的资源进行交互的库，如下所示：

import urllib.request

1. 接下来要做的是使用这个request库从互联网上拉取一些资源，如下面的代码所示：

f = urllib.request.urlretrieve("https://archive.ics.uci.edu/ml/machine-learning-databases/kddcup99-mld/kddcup.data.gz"),"kddcup.data.gz"

这个命令将需要一些时间来处理。一旦文件被下载，我们可以看到 Python 已经返回，控制台是活动的。

1. 接下来，使用SparkContext加载这个。所以，在 Python 中，SparkContext被实例化或对象化为sc变量，如下所示：

sc

此输出如下面的代码片段所示：

SparkContext
Spark UI
Version
 v2.3.3
Master
 local[*]
AppName
 PySparkShell

将数据导入 Spark

1. 接下来，使用sc将 KDD cup 数据加载到 PySpark 中，如下面的命令所示：

raw_data = sc.textFile("./kddcup.data.gz")

1. 在下面的命令中，我们可以看到原始数据现在在raw_data变量中：

raw_data

此输出如下面的代码片段所示：

./kddcup.data,gz MapPartitionsRDD[3] at textFile at NativeMethodAccessorImpl.java:0

如果我们输入raw_data变量，它会给我们关于kddcup.data.gz的详细信息，其中包含数据文件的原始数据，并告诉我们关于MapPartitionsRDD。

现在我们知道如何将数据加载到 Spark 中，让我们学习一下如何使用 Spark RDD 进行并行化。

使用 Spark RDD 进行并行化

现在我们知道如何在从互联网接收的文本文件中创建 RDD，我们可以看一种不同的创建这个 RDD 的方法。让我们讨论一下如何使用我们的 Spark RDD 进行并行化。

在这一部分，我们将涵盖以下主题：

• 什么是并行化？

• 我们如何将 Spark RDD 并行化？

让我们从并行化开始。

什么是并行化？

了解 Spark 或任何语言的最佳方法是查看文档。如果我们查看 Spark 的文档，它清楚地说明，对于我们上次使用的textFile函数，它从 HDFS 读取文本文件。

另一方面，如果我们看一下parallelize的定义，我们可以看到这是通过分发本地 Scala 集合来创建 RDD。

使用parallelize创建 RDD 和使用textFile创建 RDD 之间的主要区别在于数据的来源。

让我们看看这是如何实际工作的。让我们回到之前离开的 PySpark 安装屏幕。因此，我们导入了urllib，我们使用urllib.request从互联网检索一些数据，然后我们使用SparkContext和textFile将这些数据加载到 Spark 中。另一种方法是使用parallelize。

让我们看看我们可以如何做到这一点。让我们首先假设我们的数据已经在 Python 中，因此，为了演示目的，我们将创建一个包含一百个数字的 Python 列表如下：

a = range(100)
a

这给我们以下输出：

range(0, 100)

例如，如果我们看一下a，它只是一个包含 100 个数字的列表。如果我们将其转换为list，它将显示我们的 100 个数字的列表：

list (a)

这给我们以下输出：

[0,
 1,
 2,
 3,
 4,
 5,
 6,
 7,
 8,
 9,
 10,
 11,
 12,
 13,
 14,
 15,
 16,
 17,
 18,
 19,
 20,
 21,
 22,
 23,
 24,
 25,
 26,
 27,
...

以下命令向我们展示了如何将其转换为 RDD：

list_rdd = sc.parallelize(a)

如果我们看一下list_rdd包含什么，我们可以看到它是PythonRDD.scala:52，因此，这告诉我们 Scala 支持的 PySpark 实例已经识别出这是一个由 Python 创建的 RDD，如下所示：

list_rdd

这给我们以下输出：

PythonRDD[3] at RDD at PythonRDD.scala:52

现在，让我们看看我们可以用这个列表做什么。我们可以做的第一件事是通过以下命令计算list_rdd中有多少元素：

list_rdd.count()

这给我们以下输出：

100

我们可以看到list_rdd计数为 100。如果我们再次运行它而不切入结果，我们实际上可以看到，由于 Scala 在遍历 RDD 时是实时运行的，它比只运行a的长度要慢，后者是瞬时的。

然而，RDD 需要一些时间，因为它需要时间来遍历列表的并行化版本。因此，在小规模的情况下，只有一百个数字，可能没有这种权衡非常有帮助，但是对于更大量的数据和数据元素的更大个体大小，这将更有意义。

我们还可以从列表中取任意数量的元素，如下所示：

list_rdd.take(10)

这给我们以下输出：

[0, 1, 2, 3, 4, 5, 6, 7, 8, 9]

当我们运行上述命令时，我们可以看到 PySpark 在返回列表的前十个元素之前进行了一些计算。请注意，所有这些现在都由 PySpark 支持，并且我们正在使用 Spark 的功能来操作这个包含 100 个项目的列表。

现在让我们在list_rdd中使用reduce函数，或者在 RDDs 中一般使用，来演示我们可以用 PySpark 的 RDDs 做什么。我们将两个参数函数应用为匿名的lambda函数到reduce调用如下：

list_rdd.reduce(lambda a, b: a+b)

在这里，lambda接受两个参数a和b。它简单地将这两个数字相加，因此a+b，并返回输出。通过RDD的reduce调用，我们可以依次将 RDD 列表的前两个数字相加，返回结果，然后将第三个数字添加到结果中，依此类推。因此，最终，通过使用reduce，您可以将所有 100 个数字添加到相同的结果中。

现在，在通过分布式数据库进行一些工作之后，我们现在可以看到，从0到99的数字相加得到4950，并且所有这些都是使用 PySpark 的 RDD 方法完成的。您可能会从 MapReduce 这个术语中认出这个函数，确实，它就是这样。

我们刚刚学习了在 PySpark 中并行化是什么，以及我们如何可以并行化 Spark RDDs。这实际上相当于我们创建 RDDs 的另一种方式，对我们非常有用。现在，让我们来看一些 RDD 操作的基础知识。

RDD 操作的基础知识

现在让我们来看一些 RDD 操作的基础知识。了解某个功能的最佳方法是查看文档，以便我们可以严格理解函数的执行方式。

这是非常重要的原因是文档是函数定义和设计用途的黄金来源。通过阅读文档，我们确保我们在理解上尽可能接近源头。相关文档的链接是spark.apache.org/docs/latest/rdd-programming-guide.html。

让我们从map函数开始。map函数通过将f函数应用于此 RDD 的每个元素来返回一个 RDD。换句话说，它的工作方式与我们在 Python 中看到的map函数相同。另一方面，filter函数返回一个仅包含满足谓词的元素的新 RDD，该谓词是一个布尔值，通常由输入filter函数的f函数返回。同样，这与 Python 中的filter函数非常相似。最后，collect()函数返回一个包含此 RDD 中所有元素的列表。这就是我认为阅读文档真正发光的地方，当我们看到这样的说明时。如果你只是在谷歌搜索这个，这种情况永远不会出现在 Stack Overflow 或博客文章中。

因此，我们说collect()只有在预期结果数组很小的情况下才应该使用，因为所有数据都加载在驱动程序的内存中。这意味着，如果我们回想一下第一章，安装 PySpark 并设置开发环境，Spark 非常出色，因为它可以在许多不同的独立机器上收集和并行化数据，并且可以从一个终端透明地操作。collect()的说明是，如果我们调用collect()，则生成的 RDD 将完全加载到驱动程序的内存中，在这种情况下，我们将失去在 Spark 实例集群中分发数据的好处。

现在我们知道了所有这些，让我们看看如何实际将这三个函数应用于我们的数据。因此，返回到 PySpark 终端；我们已经将原始数据作为文本文件加载，就像我们在之前的章节中看到的那样。

我们将编写一个filter函数来查找所有包含单词normal的行，指示 RDD 数据，如下面的屏幕截图所示：

contains_normal = raw_data.filter(lambda line: "normal." in line)

让我们分析一下这意味着什么。首先，我们正在为 RDD 原始数据调用filter函数，并且我们正在向其提供一个匿名的lambda函数，该函数接受一个line参数并返回谓词，正如我们在文档中所读到的，关于单词normal是否存在于该行中。此刻，正如我们在之前的章节中讨论的那样，我们实际上还没有计算这个filter操作。我们需要做的是调用一个实际整合数据并迫使 Spark 计算某些内容的函数。在这种情况下，我们可以依赖contains_normal，就像下面的屏幕截图中所示的那样：

您可以看到，在原始数据中，包含单词normal的行数超过了 970,000 行。要使用filter函数，我们提供了一个lambda函数，并使用一个整合函数，比如counts，来强制 Spark 计算和计算底层 DataFrame 中的数据。

对于第二个例子，我们将使用 map。由于我们下载了 KDD 杯数据，我们知道它是一个逗号分隔的值文件，因此，我们很容易做的一件事是通过两个逗号拆分每一行，如下所示：

split_file = raw_data.map(lambda line: line.split(","))

让我们分析一下发生了什么。我们在raw_data上调用map函数。我们向它提供了一个名为line的匿名lambda函数，在这个函数中，我们使用,来分割line函数。结果是一个分割文件。现在，这里真正发挥了 Spark 的力量。回想一下，在contains_normal.过滤器中，当我们调用一个强制 Spark 计算count的函数时，需要几分钟才能得出正确的结果。如果我们执行map函数，它会产生相同的效果，因为我们需要对数百万行数据进行映射。因此，快速预览我们的映射函数是否正确运行的一种方法是，我们可以将几行材料化而不是整个文件。

为了做到这一点，我们可以使用之前使用过的take函数，如下面的截图所示：

这可能需要几秒钟，因为我们只取了五行，这是我们的分割，实际上相当容易管理。如果我们查看这个样本输出，我们可以理解我们的map函数已经成功创建。我们可以做的最后一件事是在原始数据上调用collect()，如下所示：

raw_data.collect()

这旨在将 Spark 的 RDD 数据结构中的所有原始数据移动到内存中。

总结

在本章中，我们学习了如何在 Spark RDD 上加载数据，还介绍了 Spark RDD 的并行化。在加载数据之前，我们简要概述了 UCI 机器学习存储库。我们概述了基本的 RDD 操作，并检查了官方文档中的函数。

在下一章中，我们将介绍大数据清洗和数据整理。

第三章：使用 Spark 笔记本进行大数据清洗和整理

在本章中，我们将学习使用 Spark 笔记本进行大数据清洗和整理。我们还将看看在笔记本应用程序上使用 Spark 如何有效地使用 RDD。我们将使用 Spark 笔记本快速迭代想法，并进行抽样/过滤 RDD 以挑选出相关数据点。我们还将学习如何拆分数据集并使用集合操作创建新的组合。

在本章中，我们将讨论以下主题：

• 使用 Spark 笔记本快速迭代想法

• 对 RDD 进行抽样/过滤以挑选出相关数据点

• 拆分数据集并创建一些新的组合

使用 Spark 笔记本快速迭代想法

在这一部分，我们将回答以下问题：

• 什么是 Spark 笔记本？

• 如何启动 Spark 笔记本？

• 如何使用 Spark 笔记本？

让我们从为 Spark 设置类似 Jupyter Notebook 的环境开始。Spark 笔记本只是一个使用 Scala 和 Spark 的交互式和反应式数据科学环境。

如果我们查看 GitHub 页面（github.com/spark-notebook/spark-notebook），我们会发现笔记本的功能实际上非常简单，如下截图所示：

如果我们看一下 Spark 笔记本，我们会发现它们看起来非常像 Python 开发人员使用的 Jupyter 笔记本。您可以在文本框中输入一些代码，然后在文本框下方执行代码，这与笔记本格式类似。这使我们能够使用 Apache Spark 和大数据生态系统执行可重现的分析。

因此，我们可以直接使用 Spark 笔记本，我们只需要转到 Spark 笔记本网站，然后点击“快速启动”即可启动笔记本，如下截图所示：

我们需要确保我们正在运行 Java 7。我们可以看到设置步骤也在文档中提到，如下截图所示：

Spark 笔记本的主要网站是spark-notebook.io，在那里我们可以看到许多选项。以下截图显示了其中一些选项：

我们可以下载 TAR 文件并解压缩。您可以使用 Spark 笔记本，但是在本书中我们将使用 Jupyter Notebook。因此，回到 Jupyter 环境，我们可以查看 PySpark 附带的代码文件。在第三章笔记本中，我们已经包含了一个方便的方法来设置环境变量，以使 PySpark 与 Jupyter 一起工作，如下截图所示：

首先，我们需要在我们的环境中创建两个新的环境变量。如果您使用 Linux，可以使用 Bash RC。如果您使用 Windows，您只需要更改和编辑系统环境变量。有多个在线教程可以帮助您完成此操作。我们要做的是编辑或包含PYSPARK_DRIVER_PYTHON变量，并将其指向您的 Jupyter Notebook 安装位置。如果您使用 Anaconda，可能会指向 Anaconda Jupyter Bash 文件。由于我们使用的是 WinPython，我已将其指向了我的 WinPython Jupyter Notebook Bash 文件。我们要导出的第二个环境变量只是PYSPARK_DRIVER_PYTHON_OPTS。

其中一个建议是，我们在选项中包括笔记本文件夹和笔记本应用程序，要求它不要在浏览器中打开，并告诉它要绑定到哪个端口。 在实践中，如果您使用的是 Windows 和 WinPython 环境，那么您实际上不需要在这里使用这行代码，您可以直接跳过它。 完成后，只需从命令行重新启动 PySpark。 发生的情况是，与我们以前看到的控制台不同，它直接启动到 Jupyter Notebook 实例，并且我们可以像在 Jupyter Notebook 中一样使用 Spark 和 SparkContext 变量。 因此，让我们测试一下，如下所示：

sc

我们立即获得了我们的SparkContext，告诉我们 Spark 的版本是2.3.3，我们的Master是local，AppName是 Python SparkShell（PySparkShell），如下面的代码片段所示：

SparkContext
Spark UI
Version
 v2.3.3
Master
 local[*]
AppName
 PySparkShell

因此，现在我们知道了如何在 Jupyter 中创建类似笔记本的环境。 在下一节中，我们将看一下对 RDD 进行抽样和过滤以挑选出相关数据点。

抽样/过滤 RDD 以挑选出相关数据点

在本节中，我们将查看对 RDD 进行抽样和过滤以挑选出相关数据点。 这是一个非常强大的概念，它使我们能够规避大数据的限制，并在特定样本上执行我们的计算。

现在让我们检查抽样不仅加速了我们的计算，而且还给了我们对我们试图计算的统计量的良好近似。 为此，我们首先导入time库，如下所示：

from time import time

我们接下来要做的是查看 KDD 数据库中包含单词normal的行或数据点：

raw_data = sc.textFile("./kdd.data.gz")

我们需要创建raw_data的样本。 我们将样本存储到sample变量中，我们正在从raw_data中进行无替换的抽样。 我们正在抽样数据的 10％，并且我们提供42作为我们的随机种子：

sampled = raw_data.sample(False, 0.1, 42)

接下来要做的是链接一些map和filter函数，就像我们通常处理未抽样数据集一样：

contains_normal_sample = sampled.map(lambda x: x.split(",")).filter(lambda x: "normal" in x)

接下来，我们需要计算在样本中计算行数需要多长时间：

t0 = time()
num_sampled = contains_normal_sample.count()
duration = time() - t0

我们在这里发布计数声明。 正如您从上一节中所知，这将触发 PySpark 中contains_normal_sample中定义的所有计算，并且我们记录了样本计数发生之前的时间。 我们还记录了样本计数发生后的时间，这样我们就可以看到在查看样本时需要多长时间。 一旦完成了这一点，让我们来看看以下代码片段中duration持续了多长时间：

duration

输出将如下所示：

23.724565505981445

我们花了 23 秒来运行这个操作，占数据的 10％。 现在，让我们看看如果我们在所有数据上运行相同的转换会发生什么：

contains_normal = raw_data.map(lambda x: x.split(",")).filter(lambda x: "normal" in x)
t0 = time()
num_sampled = contains_normal.count()
duration = time() - t0

让我们再次看一下duration：

duration 

这将提供以下输出：

36.51565098762512

有一个小差异，因为我们正在比较36.5秒和23.7秒。 但是，随着数据集变得更加多样化，以及您处理的数据量变得更加复杂，这种差异会变得更大。 这其中的好处是，如果您通常处理大数据，使用数据的小样本验证您的答案是否合理可以帮助您更早地捕捉错误。

最后要看的是我们如何使用takeSample。 我们只需要使用以下代码：

data_in_memory = raw_data.takeSample(False, 10, 42)

正如我们之前学到的，当我们呈现新函数时，我们调用takeSample，它将给我们10个具有随机种子42的项目，现在我们将其放入内存。 现在这些数据在内存中，我们可以使用本机 Python 方法调用相同的map和filter函数，如下所示：

contains_normal_py = [line.split(",") for line in data_in_memory if "normal" in line]
len(contains_normal_py)

输出将如下所示：

1

我们现在通过将data_in_memory带入来计算我们的contains_normal函数。 这很好地说明了 PySpark 的强大之处。

我们最初抽取了 10,000 个数据点的样本，这导致了机器崩溃。 因此，在这里，我们将取这十个数据点，看看它是否包含单词normal。

我们可以看到在前一个代码块中计算已经完成，它比在 PySpark 中进行计算花费了更长的时间并且使用了更多的内存。这就是为什么我们使用 Spark，因为 Spark 允许我们并行处理任何大型数据集，并且以并行方式操作它，这意味着我们可以用更少的内存和更少的时间做更多的事情。在下一节中，我们将讨论拆分数据集并使用集合操作创建新的组合。

拆分数据集并创建一些新的组合

在本节中，我们将看看如何拆分数据集并使用集合操作创建新的组合。我们将学习特别是减法和笛卡尔积。

让我们回到我们一直在查看包含单词normal的数据集中的行的 Jupyter 笔记本的第三章。让我们尝试获取不包含单词normal的所有行。一种方法是使用filter函数查看不包含normal的行。但是，在 PySpark 中我们可以使用一些不同的东西：一个名为subtract的函数来取整个数据集并减去包含单词normal的数据。让我们看看以下片段：

normal_sample = sampled.filter(lambda line: "normal." in line)

然后我们可以通过从整个样本中减去normal样本来获得不包含单词normal的交互或数据点如下：

non_normal_sample = sampled.subtract(normal_sample)

我们取normal样本，然后从整个样本中减去它，这是整个数据集的 10%。让我们按如下方式发出一些计数：

sampled.count()

这将为我们提供以下输出：

490705

正如你所看到的，数据集的 10%给我们490705个数据点，其中有一些包含单词normal的数据点。要找出它的计数，写下以下代码：

normal_sample.count()

这将为我们提供以下输出：

97404

所以，这里有97404个数据点。如果我们计算正常样本，因为我们只是从另一个样本中减去一个样本，计数应该大约略低于 400,000 个数据点，因为我们有 490,000 个数据点减去 97,000 个数据点，这应该导致大约 390,000。让我们看看使用以下代码片段会发生什么：

non_normal_sample.count()

这将为我们提供以下输出：

393301

正如预期的那样，它返回了393301的值，这验证了我们的假设，即减去包含normal的数据点会给我们所有非正常的数据点。

现在让我们讨论另一个名为cartesian的函数。这允许我们给出两个不同特征的不同值之间的所有组合。让我们看看以下代码片段中这是如何工作的：

feature_1 = sampled.map(lambda line: line.split(",")).map(lambda features: features[1]).distinct()

在这里，我们使用,来拆分line函数。因此，我们将拆分逗号分隔的值 - 对于拆分后得到的所有特征，我们取第一个特征，并找到该列的所有不同值。我们可以重复这个过程来获取第二个特征，如下所示：

feature_2 = sampled.map(lambda line: line.split(",")).map(lambda features: features[2]).distinct()

因此，我们现在有两个特征。我们可以查看feature_1和feature_2中的实际项目，如下所示，通过发出我们之前看到的collect()调用：

f1 = feature_1.collect()
f2 = feature_2.collect()

让我们分别看一下如下：

f1

这将提供以下结果：

['tcp', 'udp', 'icmp']

所以，f1有三个值；让我们检查f2如下：

f2

这将为我们提供以下输出：

f2有更多的值，我们可以使用cartesian函数收集f1和f2之间的所有组合如下：

len(feature_1.cartesian(feature_2).collect())

这将为我们提供以下输出：

198

这是我们如何使用cartesian函数找到两个特征之间的笛卡尔积。在本章中，我们看了 Spark 笔记本；抽样、过滤和拆分数据集；以及使用集合操作创建新的组合。

摘要

在本章中，我们看了 Spark 笔记本进行快速迭代。然后我们使用抽样或过滤来挑选出相关的数据点。我们还学会了如何拆分数据集并使用集合操作创建新的组合。

在下一章中，我们将介绍将数据聚合和汇总为有用的报告。

第四章：将数据聚合和汇总为有用的报告

在本章中，我们将学习如何将数据聚合和汇总为有用的报告。我们将学习如何使用 map 和 reduce 函数计算平均值，执行更快的平均计算，并使用键值对数据点的数据透视表。

本章中，我们将涵盖以下主题：

• 使用 map 和 reduce 计算平均值

• 使用聚合进行更快的平均计算

• 使用键值对数据点进行数据透视表

使用 map 和 reduce 计算平均值

在本节中，我们将回答以下三个主要问题：

• 我们如何计算平均值？

• 什么是 map？

• 什么是 reduce？

您可以在spark.apache.org/docs/latest/api/python/pyspark.html?highlight=map#pyspark.RDD.map上查看文档。

map 函数接受两个参数，其中一个是可选的。map 的第一个参数是 f，它是一个应用于整个 RDD 的函数。第二个参数或参数是 preservesPartitioning 参数，默认值为 False。

如果我们查看文档，它说 map 通过将函数应用于此 RDD 的每个元素来简单地返回一个新的 RDD，显然，此函数指的是我们输入到 map 函数本身的 f。文档中有一个非常简单的例子，如果我们并行化一个包含三个字符 b、a 和 c 的 rdd 方法，并且我们映射一个创建每个元素的元组的函数，那么我们将创建一个包含三个元组的列表，其中原始字符放在元组的第一个元素中，整数 1 放在第二个元素中，如下所示：

rdd =  sc.paralleize(["b", "a", "c"])
sorted(rdd.map(lambda x: (x, 1)).collect())

这将给我们以下输出：

[('a', 1), ('b', 1), ('c', 1)]

reduce 函数只接受一个参数，即 f。f 是一个将列表减少为一个数字的函数。从技术角度来看，指定的可交换和可结合的二进制运算符减少了此 RDD 的元素。

让我们使用我们一直在使用的 KDD 数据来举个例子。我们启动我们的 Jupyter Notebook 实例，它链接到一个 Spark 实例，就像我们以前做过的那样。然后我们通过从本地磁盘加载 kddcup.data.gz 文本文件来创建一个 raw_data 变量，如下所示：

raw_data = sc.textFile("./kddcup.data.gz")

接下来要做的是将此文件拆分为 csv，然后我们将过滤包含单词 normal 的特征 41 的行：

csv = raw_data.map(lambda x: x.split(","))
normal_data = csv.filter(lambda x: x[41]=="normal.")

然后我们使用 map 函数将这些数据转换为整数，最后，我们可以使用 reduce 函数来计算 total_duration，然后我们可以打印 total_duration 如下：

duration = normal_data.map(lambda x: int(x[0]))
total_duration = duration.reduce(lambda x, y: x+y)
total_duration

然后我们将得到以下输出：

211895753

接下来要做的是将 total_duration 除以数据的计数，如下所示：

total_duration/(normal_data.count())

这将给我们以下输出：

217.82472416710442

稍微计算后，我们将使用 map 和 reduce 创建两个计数。我们刚刚学会了如何使用 PySpark 计算平均值，以及 PySpark 中的 map 和 reduce 函数是什么。

使用聚合进行更快的平均计算

在上一节中，我们看到了如何使用 map 和 reduce 计算平均值。现在让我们看看如何使用 aggregate 函数进行更快的平均计算。您可以参考前一节中提到的文档。

aggregate 是一个带有三个参数的函数，其中没有一个是可选的。

第一个是 zeroValue 参数，我们在其中放入聚合结果的基本情况。

第二个参数是顺序运算符 (seqOp)，它允许您在 zeroValue 之上堆叠和聚合值。您可以从 zeroValue 开始，将您的 RDD 中的值传递到 seqOp 函数中，并将其堆叠或聚合到 zeroValue 之上。

最后一个参数是combOp，表示组合操作，我们只需将通过seqOp参数聚合的zeroValue参数组合成一个值，以便我们可以使用它来完成聚合。

因此，我们正在聚合每个分区的元素，然后使用组合函数和中性零值对所有分区的结果进行聚合。在这里，我们有两件事需要注意：

1. op函数允许修改t1，但不应修改t2

2. 第一个函数seqOp可以返回不同的结果类型U

在这种情况下，我们都需要一个操作来将T合并到U，以及一个操作来合并这两个U。

让我们去我们的 Jupyter Notebook 检查这是如何完成的。aggregate允许我们同时计算总持续时间和计数。我们调用duration_count函数。然后我们取normal_data并对其进行聚合。请记住，聚合有三个参数。第一个是初始值；也就是零值，(0,0)。第二个是一个顺序操作，如下所示：

duration_count = duration.aggregate(
 (0,0),
 (lambda db, new_value: (db[0] + new_value, db[1] + 1))
)

我们需要指定一个具有两个参数的lambda函数。第一个参数是当前的累加器，或者聚合器，或者也可以称为数据库（db）。然后，在我们的lambda函数中，我们有第二个参数new_value，或者我们在 RDD 中处理的当前值。我们只是想对数据库做正确的事情，也就是说，我们知道我们的数据库看起来像一个元组，第一个元素是持续时间的总和，第二个元素是计数。在这里，我们知道我们的数据库看起来像一个元组，持续时间的总和是第一个元素，计数是第二个元素。每当我们查看一个新值时，我们需要将新值添加到当前的运行总数中，并将1添加到当前的运行计数中。

运行总数是第一个元素，db[0]。然后我们只需要将1添加到第二个元素db[1]，即计数。这是顺序操作。

每当我们得到一个new_value，如前面的代码块所示，我们只需将其添加到运行总数中。而且，因为我们已经将new_value添加到运行总数中，我们需要将计数增加1。其次，我们需要放入组合器操作。现在，我们只需要将两个单独的数据库db1和db2的相应元素组合如下：

duration_count = duration.aggregate(
 (0,0),
 (lambda db, new_value: (db[0] + new_value, db[1] + 1)),
 (lambda db1, db2: (db1[0] + db2[0], db1[1] + db2[1]))
)

由于持续时间计数是一个元组，它在第一个元素上收集了我们的总持续时间，在第二个元素上记录了我们查看的持续时间数量，计算平均值非常简单。我们需要将第一个元素除以第二个元素，如下所示：

duration_count[0]/duration_count[1]

这将给我们以下输出：

217.82472416710442

您可以看到它返回了与我们在上一节中看到的相同的结果，这很棒。在下一节中，我们将看一下带有键值对数据点的数据透视表。

带有键值对数据点的数据透视表

数据透视表非常简单且易于使用。我们将使用大型数据集，例如 KDD 杯数据集，并根据某些键对某些值进行分组。

例如，我们有一个包含人和他们最喜欢的水果的数据集。我们想知道有多少人把苹果作为他们最喜欢的水果，因此我们将根据水果将人数进行分组，这是值，而不是键。这就是数据透视表的简单概念。

我们可以使用map函数将 KDD 数据集移动到键值对范例中。我们使用lambda函数将数据集的特征41映射到kv键值，并将值附加如下：

kv = csv.map(lambda x: (x[41], x))
kv.take(1)

我们使用特征41作为键，值是数据点，即x。我们可以使用take函数来获取这些转换行中的一个，以查看其外观。

现在让我们尝试类似于前面的例子。为了找出特征41中每种数值的总持续时间，我们可以再次使用map函数，简单地将41特征作为我们的键。我们可以将数据点中第一个数字的浮点数作为我们的值。我们将使用reduceByKey函数来减少每个键的持续时间。

因此，reduceByKey不仅仅是减少所有数据点，而是根据它们所属的键来减少持续时间数字。您可以在spark.apache.org/docs/latest/api/python/pyspark.html?highlight=map#pyspark.RDD.reduceByKey上查看文档。reduceByKey使用关联和交换的reduce函数合并每个键的值。它在将结果发送到减速器之前在每个映射器上执行本地合并，这类似于 MapReduce 中的组合器。

reduceByKey函数只需一个参数。我们将使用lambda函数。我们取两个不同的持续时间并将它们相加，PySpark 足够聪明，可以根据键应用这个减少函数，如下所示：

kv_duration = csv.map(lambda x: (x[41], float(x[0]))).reduceByKey(lambda x, y: x+y)
kv_duration.collect()

结果输出如下截图所示：

如果我们收集键值持续时间数据，我们可以看到持续时间是由出现在特征41中的值收集的。如果我们在 Excel 中使用数据透视表，有一个方便的函数是countByKey函数，它执行的是完全相同的操作，如下所示：

kv.countByKey()

这将给我们以下输出：

您可以看到调用kv.countByKey()函数与调用reduceByKey函数相同，先前是从键到持续时间的映射。

摘要

在本章中，我们学习了如何使用map和reduce计算平均值。我们还学习了使用aggregate进行更快的平均计算。最后，我们了解到数据透视表允许我们根据特征的不同值对数据进行聚合，并且在 PySpark 中，我们可以利用reducedByKey或countByKey等方便的函数。

在下一章中，我们将学习关于 MLlib 的内容，其中涉及机器学习，这是一个非常热门的话题。

第五章：使用 MLlib 进行强大的探索性数据分析

在本章中，我们将探索 Spark 执行回归任务的能力，使用线性回归和支持向量机等模型。我们将学习如何使用 MLlib 计算汇总统计，并使用 Pearson 和 Spearman 相关性发现数据集中的相关性。我们还将在大型数据集上测试我们的假设。

我们将涵盖以下主题：

• 使用 MLlib 计算汇总统计

• 使用 Pearson 和 Spearman 方法发现相关性

• 在大型数据集上测试我们的假设

使用 MLlib 计算汇总统计

在本节中，我们将回答以下问题：

• 什么是汇总统计？

• 我们如何使用 MLlib 创建汇总统计？

MLlib 是随 Spark 一起提供的机器学习库。最近有一个新的发展，允许我们使用 Spark 的数据处理能力传输到 Spark 本身的机器学习能力。这意味着我们不仅可以使用 Spark 来摄取、收集和转换数据，还可以分析和使用它来构建 PySpark 平台上的机器学习模型，这使我们能够拥有更无缝的可部署解决方案。

汇总统计是一个非常简单的概念。我们熟悉某个变量的平均值、标准差或方差。这些是数据集的汇总统计。之所以称其为汇总统计，是因为它通过某个统计量给出了某个东西的摘要。例如，当我们谈论数据集的平均值时，我们正在总结数据集的一个特征，而这个特征就是平均值。

让我们看看如何在 Spark 中计算汇总统计。关键因素在于colStats函数。colStats函数计算rdd输入的逐列汇总统计。colStats函数接受一个参数，即rdd，并允许我们使用 Spark 计算不同的汇总统计。

让我们看一下 Jupyter Notebook 中的代码（可在github.com/PacktPublishing/Hands-On-Big-Data-Analytics-with-PySpark/tree/master/Chapter05找到），在Chapter5.ipynb中的本章。我们将首先从kddcup.data.gz文本文件中收集数据，并将其传输到raw_data变量中：

raw_data = sc.textFile("./kddcup.data.gz")

kddcup.data文件是一个逗号分隔值（CSV）文件。我们必须通过,字符拆分这些数据，并将其放入csv变量中，如下所示：

csv = raw_data.map(lambda x: x.split(","))

让我们取数据文件的第一个特征x[0]；这个特征代表持续时间，也就是数据的方面。我们将把它转换为整数，并将其包装成列表，如下所示：

duration = csv.map(lambda x: [int(x[0])])

这有助于我们对多个变量进行汇总统计，而不仅仅是其中一个。要激活colStats函数，我们需要导入Statistics包，如下面的代码片段所示：

from pyspark.mllib.stat import Statistics

这个Statistics包是pyspark.mllib.stat的一个子包。现在，我们需要在Statistics包中调用colStats函数，并向其提供一些数据。这里，我们谈论的是数据集中的持续时间数据，并将汇总统计信息输入到summary变量中：

summary = Statistics.colStats(duration)

要访问不同的汇总统计，如平均值、标准差等，我们可以调用summary对象的函数，并访问不同的汇总统计。例如，我们可以访问mean，由于我们的持续时间数据集中只有一个特征，我们可以通过00索引对其进行索引，然后得到数据集的平均值，如下所示：

summary.mean()[0]

这将给我们以下输出：

47.97930249928637

同样，如果我们从 Python 标准库中导入sqrt函数，我们可以创建数据集中持续时间的标准差，如下面的代码片段所示：

from math import sqrt
sqrt(summary.variance()[0])

这将给我们以下输出：

707.746472305374

如果我们不使用[0]对摘要统计信息进行索引，我们可以看到summary.max()和summary.min()会返回一个数组，其中第一个元素是我们所需的摘要统计信息，如下面的代码片段所示：

summary.max()
array ([58329.]) #output
summary.min()
array([0.])  #output

使用 Pearson 和 Spearman 相关性来发现相关性

在这一部分，我们将看到在数据集中计算相关性的两种不同方法，这两种方法分别称为 Pearson 和 Spearman 相关性。

Pearson 相关性

Pearson 相关系数向我们展示了两个不同变量同时变化的程度，然后根据它们的变化程度进行调整。如果你有一个数据集，这可能是计算相关性最流行的方法之一。

Spearman 相关性

Spearman 秩相关不是内置在 PySpark 中的默认相关计算，但它非常有用。Spearman 相关系数是排名变量之间的 Pearson 相关系数。使用不同的相关性观察方法可以让我们更全面地理解相关性的工作原理。让我们看看在 PySpark 中如何计算这个。

计算 Pearson 和 Spearman 相关性

为了理解这一点，让我们假设我们正在从数据集中取出前三个数值变量。为此，我们要访问之前定义的csv变量，我们只需使用逗号（,）分割raw_data。我们只考虑前三列是数值的特征。我们不会取包含文字的任何内容；我们只对纯粹基于数字的特征感兴趣。在我们的例子中，在kddcup.data中，第一个特征的索引是0；特征 5 和特征 6 的索引分别是4和5，这些是我们拥有的数值变量。我们使用lambda函数将这三个变量放入一个列表中，并将其放入metrics变量中：

metrics = csv.map(lambda x: [x[0], x[4], x[5]])
Statistics.corr(metrics, method="spearman")

这将给我们以下输出：

array([[1\.       ,  0.01419628,  0.29918926],
 [0.01419628,  1\.        , -0.16793059],
 [0.29918926, -0.16793059,  1\.        ]])

在使用 MLlib 计算摘要统计信息部分，我们只是将第一个特征放入一个列表中，并创建了一个长度为 1 的列表。在这里，我们将三个变量的三个量放入同一个列表中。现在，每个列表的长度都是三。

为了计算相关性，我们在metrics变量上调用corr方法，并指定method为"spearman"。PySpark 会给我们一个非常简单的矩阵，告诉我们变量之间的相关性。在我们的例子中，metrics变量中的第三个变量比第二个变量更相关。

如果我们再次在metrics上运行corr，但指定方法为pearson，那么它会给我们 Pearson 相关性。因此，让我们看看为什么我们需要有资格称为数据科学家或机器学习研究人员来调用这两个简单的函数，并简单地改变第二个参数的值。许多机器学习和数据科学都围绕着我们对统计学的理解，对数据行为的理解，对机器学习模型基础的理解以及它们的预测能力是如何产生的。

因此，作为一个机器学习从业者或数据科学家，我们只是把 PySpark 当作一个大型计算器来使用。当我们使用计算器时，我们从不抱怨计算器使用简单——事实上，它帮助我们以更直接的方式完成目标。PySpark 也是一样的情况；一旦我们从数据工程转向 MLlib，我们会注意到代码变得逐渐更容易。它试图隐藏数学的复杂性，但我们需要理解不同相关性之间的差异，也需要知道如何以及何时使用它们。

在大型数据集上测试我们的假设

在本节中，我们将研究假设检验，并学习如何使用 PySpark 测试假设。让我们看看 PySpark 中实现的一种特定类型的假设检验。这种假设检验称为 Pearson 卡方检验。卡方检验评估了两个数据集之间的差异是由偶然因素引起的可能性有多大。

例如，如果我们有一个没有任何人流量的零售店，突然之间有了人流量，那么这是随机发生的可能性有多大，或者现在我们得到的访客水平与以前相比是否有任何统计学上显著的差异？之所以称之为卡方检验，是因为测试本身参考了卡方分布。您可以参考在线文档了解更多关于卡方分布的信息。

Pearson 的卡方检验有三种变体。我们将检查观察到的数据集是否与理论数据集分布不同。

让我们看看如何实现这一点。让我们从pyspark.mllib.linalg中导入Vectors包开始。使用这个向量，我们将创建一个存储中每天访客频率的密集向量。

假设访问频率从每小时的 0.13 到 0.61，0.8，0.5，最后在星期五结束时为 0.3。因此，我们将这些访客频率放入visitors_freq变量中。由于我们使用 PySpark，我们可以很容易地从Statistics包中运行卡方检验，我们已经导入如下：

from pyspark.mllib.linalg import Vectors
visitors_freq = Vectors.dense(0.13, 0.61, 0.8, 0.5, 0.3)
print(Statistics.chiSqTest(visitors_freq))

通过运行卡方检验，visitors_freq变量为我们提供了大量有用的信息，如下截图所示：

前面的输出显示了卡方检验的摘要。我们使用了pearson方法，在我们的 Pearson 卡方检验中有 4 个自由度，统计数据为 0.585，这意味着pValue为 0.964。这导致没有反对零假设的推定。这样，观察到的数据遵循与预期相同的分布，这意味着我们的访客实际上并没有不同。这使我们对假设检验有了很好的理解。

摘要

在本章中，我们学习了摘要统计信息并使用 MLlib 计算摘要统计信息。我们还了解了 Pearson 和 Spearman 相关性，以及如何使用 PySpark 在数据集中发现这些相关性。最后，我们学习了一种特定的假设检验方法，称为 Pearson 卡方检验。然后，我们使用 PySpark 的假设检验函数在大型数据集上测试了我们的假设。

在下一章中，我们将学习如何在 Spark SQL 中处理大数据的结构。

第六章：使用 SparkSQL 为您的大数据添加结构

在本章中，我们将学习如何使用 Spark SQL 模式操作数据框，并使用 Spark DSL 构建结构化数据操作的查询。到目前为止，我们已经学会了将大数据导入 Spark 环境使用 RDD，并对这些大数据进行多个操作。现在让我们看看如何操作我们的数据框并构建结构化数据操作的查询。

具体来说，我们将涵盖以下主题：

• 使用 Spark SQL 模式操作数据框

• 使用 Spark DSL 构建查询

使用 Spark SQL 模式操作数据框

在本节中，我们将学习更多关于数据框，并学习如何使用 Spark SQL。

Spark SQL 接口非常简单。因此，去除标签意味着我们处于无监督学习领域。此外，Spark 对聚类和降维算法有很好的支持。通过使用 Spark SQL 为大数据赋予结构，我们可以有效地解决学习问题。

让我们看一下我们将在 Jupyter Notebook 中使用的代码。为了保持一致，我们将使用相同的 KDD 杯数据：

1. 我们首先将textFile输入到raw_data变量中，如下所示：

raw_data = sc.textFile("./kddcup.data.gz")

1. 新的是我们从pyspark.sql中导入了两个新包：

• Row

• SQLContext

1. 以下代码向我们展示了如何导入这些包：

from pyspark.sql import Row, SQLContext
sql_context = SQLContext(sc)
csv = raw_data.map(lambda l: l.split(","))

使用SQLContext，我们创建一个新的sql_context变量，其中包含由 PySpark 创建的SQLContext变量的对象。由于我们使用SparkContext来启动这个SQLContext变量，我们需要将sc作为SQLContext创建者的第一个参数。之后，我们需要取出我们的raw_data变量，并使用l.splitlambda 函数将其映射为一个包含我们的逗号分隔值（CSV）的对象。

1. 我们将利用我们的新重要Row对象来创建一个新对象，其中定义了标签。这是为了通过我们正在查看的特征对我们的数据集进行标记，如下所示：

rows = csv.map(lambda p: Row(duration=int(p[0]), protocol=p[1], service=p[2]))

在上面的代码中，我们取出了我们的逗号分隔值（csv），并创建了一个Row对象，其中包含第一个特征称为duration，第二个特征称为protocol，第三个特征称为service。这直接对应于实际数据集中的标签。

1. 现在，我们可以通过在sql_context变量中调用createDataFrame函数来创建一个新的数据框。要创建这个数据框，我们需要提供我们的行数据对象，结果对象将是df中的数据框。之后，我们需要注册一个临时表。在这里，我们只是称之为rdd。通过这样做，我们现在可以使用普通的 SQL 语法来查询由我们的行构造的临时表中的内容：

df = sql_context.createDataFrame(rows)
df.registerTempTable("rdd")

1. 在我们的示例中，我们需要从rdd中选择duration，这是一个临时表。我们在这里选择的协议等于'tcp'，而我们在一行中的第一个特征是大于2000的duration，如下面的代码片段所示：

sql_context.sql("""SELECT duration FROM rdd WHERE protocol = 'tcp' AND duration > 2000""")

1. 现在，当我们调用show函数时，它会给我们每个符合这些条件的数据点：

sql_context.sql("""SELECT duration FROM rdd WHERE protocol = 'tcp' AND duration > 2000""").show()

1. 然后我们将得到以下输出：

+--------+
|duration|
+--------+
|   12454|
|   10774|
|   13368|
|   10350|
|   10409|
|   14918|
|   10039|
|   15127|
|   25602|
|   13120|
|    2399|
|    6155|
|   11155|
|   12169|
|   15239|
|   10901|
|   15182|
|    9494|
|    7895|
|   11084|
+--------+
only showing top 20 rows

使用前面的示例，我们可以推断出我们可以使用 PySpark 包中的SQLContext变量将数据打包成 SQL 友好格式。

因此，PySpark 不仅支持使用 SQL 语法查询数据，还可以使用 Spark 领域特定语言（DSL）构建结构化数据操作的查询。

使用 Spark DSL 构建查询

在本节中，我们将使用 Spark DSL 构建结构化数据操作的查询：

1. 在以下命令中，我们使用了与之前相同的查询；这次使用了 Spark DSL 来说明和比较使用 Spark DSL 与 SQL 的不同之处，但实现了与我们在前一节中展示的 SQL 相同的目标：

df.select("duration").filter(df.duration>2000).filter(df.protocol=="tcp").show()

在这个命令中，我们首先取出了在上一节中创建的df对象。然后我们通过调用select函数并传入duration参数来选择持续时间。

1. 接下来，在前面的代码片段中，我们两次调用了filter函数，首先使用df.duration，第二次使用df.protocol。在第一种情况下，我们试图查看持续时间是否大于2000，在第二种情况下，我们试图查看协议是否等于"tcp"。我们还需要在命令的最后附加show函数，以获得与以下代码块中显示的相同结果。

+--------+
|duration|
+--------+
|   12454|
|   10774|
|   13368|
|   10350|
|   10409|
|   14918|
|   10039|
|   15127|
|   25602|
|   13120|
|    2399|
|    6155|
|   11155|
|   12169|
|   15239|
|   10901|
|   15182|
|    9494|
|    7895|
|   11084|
+--------+
only showing top 20 rows

在这里，我们再次有了符合代码描述的前 20 行数据点的结果。

总结

在本章中，我们涵盖了 Spark DSL，并学习了如何构建查询。我们还学习了如何使用 Spark SQL 模式操纵 DataFrames，然后我们使用 Spark DSL 构建了结构化数据操作的查询。现在我们对 Spark 有了很好的了解，让我们在接下来的章节中看一些 Apache Spark 中的技巧和技术。

在下一章中，我们将看一下 Apache Spark 程序中的转换和操作。

第七章：转换和操作

转换和操作是 Apache Spark 程序的主要构建模块。在本章中，我们将看一下 Spark 转换来推迟计算，然后看一下应该避免哪些转换。然后，我们将使用reduce和reduceByKey方法对数据集进行计算。然后，我们将执行触发实际计算的操作。在本章结束时，我们还将学习如何重用相同的rdd进行不同的操作。

在本章中，我们将涵盖以下主题：

• 使用 Spark 转换来推迟计算到以后的时间

• 避免转换

• 使用reduce和reduceByKey方法来计算结果

• 执行触发实际计算我们的有向无环图（DAG）的操作

• 重用相同的rdd进行不同的操作

使用 Spark 转换来推迟计算到以后的时间

让我们首先了解 Spark DAG 的创建。我们将通过发出操作来执行 DAG，并推迟关于启动作业的决定，直到最后一刻来检查这种可能性给我们带来了什么。

让我们看一下我们将在本节中使用的代码。

首先，我们需要初始化 Spark。我们进行的每个测试都是相同的。在开始使用之前，我们需要初始化它，如下例所示：

class DeferComputations extends FunSuite {
val spark: SparkContext = SparkSession.builder().master("local[2]").getOrCreate().sparkContext

然后，我们将进行实际测试。在这里，test被称为should defer computation。它很简单，但展示了 Spark 的一个非常强大的抽象。我们首先创建一个InputRecord的rdd，如下例所示：

test("should defer computations") {
 //given
    val input = spark.makeRDD(
        List(InputRecord(userId = "A"),
            InputRecord(userId = "B")))

InputRecord是一个具有可选参数的唯一标识符的案例类。

如果我们没有提供它和必需的参数userId，它可以是一个随机的uuid。InputRecord将在本书中用于测试目的。我们已经创建了两条InputRecord的记录，我们将对其应用转换，如下例所示：

//when apply transformation
val rdd = input
    .filter(_.userId.contains("A"))
    .keyBy(_.userId)
.map(_._2.userId.toLowerCase)
//.... built processing graph lazy

我们只会过滤userId字段中包含A的记录。然后我们将其转换为keyBy(_.userId)，然后从值中提取userId并将其映射为小写。这就是我们的rdd。所以，在这里，我们只创建了 DAG，但还没有执行。假设我们有一个复杂的程序，在实际逻辑之前创建了许多这样的无环图。

Spark 的优点是直到发出操作之前不会执行，但我们可以有一些条件逻辑。例如，我们可以得到一个快速路径的执行。假设我们有shouldExecutePartOfCode()，它可以检查配置开关，或者去 REST 服务计算rdd计算是否仍然相关，如下例所示：

if (shouldExecutePartOfCode()) {
     //rdd.saveAsTextFile("") ||
     rdd.collect().toList
  } else {
    //condition changed - don't need to evaluate DAG
 }
}

我们已经使用了简单的方法进行测试，我们只是返回true，但在现实生活中，这可能是复杂的逻辑：

private def shouldExecutePartOfCode(): Boolean = {
    //domain logic that decide if we still need to calculate
    true
    }
}

在它返回true之后，我们可以决定是否要执行 DAG。如果要执行，我们可以调用rdd.collect().toList或saveAsTextFile来执行rdd。否则，我们可以有一个快速路径，并决定我们不再对输入的rdd感兴趣。通过这样做，只会创建图。

当我们开始测试时，它将花费一些时间来完成，并返回以下输出：

"C:\Program Files\Java\jdk-12\bin\java.exe" "-javaagent:C:\Program Files\JetBrains\IntelliJ IDEA 2018.3.5\lib\idea_rt.jar=50627:C:\Program Files\JetBrains\IntelliJ IDEA 2018.3.5\bin" -Dfile.encoding=UTF-8 -classpath C:\Users\Sneha\IdeaProjects\Chapter07\out\production\Chapter07 com.company.Main

Process finished with exit code 0

我们可以看到我们的测试通过了，我们可以得出它按预期工作的结论。现在，让我们看一些应该避免的转换。

避免转换

在本节中，我们将看一下应该避免的转换。在这里，我们将专注于一个特定的转换。

我们将从理解groupByAPI 开始。然后，我们将研究在使用groupBy时的数据分区，然后我们将看一下什么是 skew 分区以及为什么应该避免 skew 分区。

在这里，我们正在创建一个交易列表。UserTransaction是另一个模型类，包括userId和amount。以下代码块显示了一个典型的交易，我们正在创建一个包含五个交易的列表：

test("should trigger computations using actions") {
 //given
 val input = spark.makeRDD(
     List(
         UserTransaction(userId = "A", amount = 1001),
         UserTransaction(userId = "A", amount = 100),
         UserTransaction(userId = "A", amount = 102),
         UserTransaction(userId = "A", amount = 1),
         UserTransaction(userId = "B", amount = 13)))

我们已经为userId = "A"创建了四笔交易，为userId = "B"创建了一笔交易。

现在，让我们考虑我们想要合并特定userId的交易以获得交易列表。我们有一个input，我们正在按userId分组，如下例所示：

//when apply transformation
val rdd = input
    .groupBy(_.userId)
    .map(x => (x._1,x._2.toList))
    .collect()
    .toList

对于每个x元素，我们将创建一个元组。元组的第一个元素是一个 ID，而第二个元素是该特定 ID 的每个交易的迭代器。我们将使用toList将其转换为列表。然后，我们将收集所有内容并将其分配给toList以获得我们的结果。让我们断言结果。rdd应该包含与B相同的元素，即键和一个交易，以及A，其中有四个交易，如下面的代码所示：

//then
rdd should contain theSameElementsAs List(
    ("B", List(UserTransaction("B", 13))),
    ("A", List(
        UserTransaction("A", 1001),
        UserTransaction("A", 100),
        UserTransaction("A", 102),
        UserTransaction("A", 1))
    )
  )
 }
}

让我们开始这个测试，并检查它是否按预期行为。我们得到以下输出：

"C:\Program Files\Java\jdk-12\bin\java.exe" "-javaagent:C:\Program Files\JetBrains\IntelliJ IDEA 2018.3.5\lib\idea_rt.jar=50822:C:\Program Files\JetBrains\IntelliJ IDEA 2018.3.5\bin" -Dfile.encoding=UTF-8 -classpath C:\Users\Sneha\IdeaProjects\Chapter07\out\production\Chapter07 com.company.Main

Process finished with exit code 0

乍一看，它已经通过了，并且按预期工作。但是，为什么我们要对它进行分组的问题就出现了。我们想要对它进行分组以将其保存到文件系统或进行一些进一步的操作，例如连接所有金额。

我们可以看到我们的输入不是正常分布的，因为几乎所有的交易都是针对userId = "A"。因此，我们有一个偏斜的键。这意味着一个键包含大部分数据，而其他键包含较少的数据。当我们在 Spark 中使用groupBy时，它会获取所有具有相同分组的元素，例如在这个例子中是userId，并将这些值发送到完全相同的执行者。

例如，如果我们的执行者有 5GB 的内存，我们有一个非常大的数据集，有数百 GB，其中一个键有 90%的数据，这意味着所有数据都将传输到一个执行者，其余的执行者将获取少数数据。因此，数据将不会正常分布，并且由于非均匀分布，处理效率将不会尽可能高。

因此，当我们使用groupBy键时，我们必须首先回答为什么要对其进行分组的问题。也许我们可以在groupBy之前对其进行过滤或聚合，然后我们只会对结果进行分组，或者根本不进行分组。我们将在以下部分中研究如何使用 Spark API 解决这个问题。

使用 reduce 和 reduceByKey 方法来计算结果

在本节中，我们将使用reduce和reduceBykey函数来计算我们的结果，并了解reduce的行为。然后，我们将比较reduce和reduceBykey函数，以确定在特定用例中应该使用哪个函数。

我们将首先关注reduceAPI。首先，我们需要创建一个UserTransaction的输入。我们有用户交易A，金额为10，B的金额为1，A的金额为101。假设我们想找出全局最大值。我们对特定键的数据不感兴趣，而是对全局数据感兴趣。我们想要扫描它，取最大值，并返回它，如下例所示：

test("should use reduce API") {
    //given
    val input = spark.makeRDD(List(
    UserTransaction("A", 10),
    UserTransaction("B", 1),
    UserTransaction("A", 101)
    ))

因此，这是减少使用情况。现在，让我们看看如何实现它，如下例所示：

//when
val result = input
    .map(_.amount)
    .reduce((a, b) => if (a > b) a else b)

//then
assert(result == 101)
}

对于input，我们需要首先映射我们感兴趣的字段。在这种情况下，我们对amount感兴趣。我们将取amount，然后取最大值。

在前面的代码示例中，reduce有两个参数，a和b。一个参数将是我们正在传递的特定 Lambda 中的当前最大值，而第二个参数将是我们现在正在调查的实际值。如果该值高于到目前为止的最大状态，我们将返回a；如果不是，它将返回b。我们将遍历所有元素，最终结果将只是一个长数字。

因此，让我们测试一下，检查结果是否确实是101，如以下代码输出所示。这意味着我们的测试通过了。

"C:\Program Files\Java\jdk-12\bin\java.exe" "-javaagent:C:\Program Files\JetBrains\IntelliJ IDEA 2018.3.5\lib\idea_rt.jar=50894:C:\Program Files\JetBrains\IntelliJ IDEA 2018.3.5\bin" -Dfile.encoding=UTF-8 -classpath C:\Users\Sneha\IdeaProjects\Chapter07\out\production\Chapter07 com.company.Main

Process finished with exit code 0

现在，让我们考虑一个不同的情况。我们想找到最大的交易金额，但这次我们想根据用户来做。我们不仅想找出用户A的最大交易，还想找出用户B的最大交易，但我们希望这些事情是独立的。因此，对于相同的每个键，我们只想从我们的数据中取出最大值，如以下示例所示：

test("should use reduceByKey API") {
    //given
    val input = spark.makeRDD(
    List(
        UserTransaction("A", 10),
        UserTransaction("B", 1),
        UserTransaction("A", 101)
    )
)

要实现这一点，reduce不是一个好选择，因为它将遍历所有的值并给出全局最大值。我们在 Spark 中有关键操作，但首先，我们要为特定的元素组做这件事。我们需要使用keyBy告诉 Spark 应该将哪个 ID 作为唯一的，并且它将仅在特定的键内执行reduce函数。因此，我们使用keyBy(_.userId)，然后得到reducedByKey函数。reduceByKey函数类似于reduce，但它按键工作，因此在 Lambda 内，我们只会得到特定键的值，如以下示例所示：

    //when
    val result = input
      .keyBy(_.userId)
      .reduceByKey((firstTransaction, secondTransaction) =>
        TransactionChecker.higherTransactionAmount(firstTransaction, secondTransaction))
      .collect()
      .toList

通过这样做，我们得到第一笔交易，然后是第二笔。第一笔将是当前的最大值，第二笔将是我们正在调查的交易。我们将创建一个辅助函数，它接受这些交易并称之为higherTransactionAmount。

higherTransactionAmount函数用于获取firstTransaction和secondTransaction。请注意，对于UserTransaction类型，我们需要传递该类型。它还需要返回UserTransaction，我们不能返回不同的类型。

如果您正在使用 Spark 的reduceByKey方法，我们需要返回与input参数相同的类型。如果firstTransaction.amount高于secondTransaction.amount，我们将返回firstTransaction，因为我们返回的是secondTransaction，所以是交易对象而不是总金额。这在以下示例中显示：

object TransactionChecker {
    def higherTransactionAmount(firstTransaction: UserTransaction, secondTransaction: UserTransaction): UserTransaction = {
        if (firstTransaction.amount > secondTransaction.amount) firstTransaction else     secondTransaction
    }
}

现在，我们将收集、添加和测试交易。在我们的测试之后，我们得到了输出，对于键B，我们应该得到交易("B", 1)，对于键A，交易("A", 101)。没有交易("A", 10)，因为我们已经过滤掉了它，但我们可以看到对于每个键，我们都能找到最大值。这在以下示例中显示：

    //then
    result should contain theSameElementsAs
      List(("B", UserTransaction("B", 1)), ("A", UserTransaction("A", 101)))
  }

}

我们可以看到测试通过了，一切都如预期的那样，如以下输出所示：

"C:\Program Files\Java\jdk-12\bin\java.exe" "-javaagent:C:\Program Files\JetBrains\IntelliJ IDEA 2018.3.5\lib\idea_rt.jar=50909:C:\Program Files\JetBrains\IntelliJ IDEA 2018.3.5\bin" -Dfile.encoding=UTF-8 -classpath C:\Users\Sneha\IdeaProjects\Chapter07\out\production\Chapter07 com.company.Main

Process finished with exit code 0

在下一节中，我们将执行触发数据计算的操作。

执行触发计算的操作

Spark 有更多触发 DAG 的操作，我们应该了解所有这些，因为它们非常重要。在本节中，我们将了解 Spark 中可以成为操作的内容，对操作进行一次遍历，并测试这些操作是否符合预期。

我们已经涵盖的第一个操作是collect。除此之外，我们还涵盖了两个操作——在上一节中我们都涵盖了reduce和reduceByKey。这两种方法都是操作，因为它们返回单个结果。

首先，我们将创建我们的交易的input，然后应用一些转换，仅用于测试目的。我们将只取包含A的用户，使用keyBy_.userId，然后只取所需交易的金额，如以下示例所示：

test("should trigger computations using actions") {
     //given
     val input = spark.makeRDD(
     List(
         UserTransaction(userId = "A", amount = 1001),
         UserTransaction(userId = "A", amount = 100),
         UserTransaction(userId = "A", amount = 102),
         UserTransaction(userId = "A", amount = 1),
         UserTransaction(userId = "B", amount = 13)))

//when apply transformation
 val rdd = input
     .filter(_.userId.contains("A"))
     .keyBy(_.userId)
     .map(_._2.amount)

我们已经知道的第一个操作是rdd.collect().toList。接下来是count()，它需要获取所有的值并计算rdd中有多少值。没有办法在不触发转换的情况下执行count()。此外，Spark 中还有不同的方法，如countApprox、countApproxDistinct、countByValue和countByValueApprox。以下示例显示了rdd.collect().toList的代码：

//then
 println(rdd.collect().toList)
 println(rdd.count()) //and all count*

如果我们有一个庞大的数据集，并且近似计数就足够了，你可以使用countApprox，因为它会快得多。然后我们使用rdd.first()，但这个选项有点不同，因为它只需要取第一个元素。有时，如果你想取第一个元素并执行我们 DAG 中的所有操作，我们需要专注于这一点，并以以下方式检查它：

println(rdd.first())

此外，在rdd上，我们有foreach()，这是一个循环，我们可以传递任何函数。假定 Scala 函数或 Java 函数是 Lambda，但要执行我们结果rdd的元素，需要计算 DAG，因为从这里开始，它就是一个操作。foreach()方法的另一个变体是foreachPartition()，它获取每个分区并返回分区的迭代器。在其中，我们有一个迭代器再次进行迭代并打印我们的元素。我们还有我们的max()和min()方法，预期的是，max()取最大值，min()取最小值。但这些方法都需要隐式排序。

如果我们有一个简单的原始类型的rdd，比如Long，我们不需要在这里传递它。但如果我们不使用map()，我们需要为 Spark 定义UserTransaction的排序，以便找出哪个元素是max，哪个元素是min。这两件事需要执行 DAG，因此它们被视为操作，如下面的例子所示：

 rdd.foreach(println(_))
 rdd.foreachPartition(t => t.foreach(println(_)))
 println(rdd.max())
 println(rdd.min())

然后我们有takeOrdered()，这是一个比first()更耗时的操作，因为first()取一个随机元素。takeOrdered()需要执行 DAG 并对所有内容进行排序。当一切都排序好后，它才取出顶部的元素。

在我们的例子中，我们取num = 1。但有时，出于测试或监控的目的，我们需要只取数据的样本。为了取样，我们使用takeSample()方法并传递一个元素数量，如下面的代码所示：

 println(rdd.takeOrdered(1).toList)
 println(rdd.takeSample(false, 2).toList)
 }
}

现在，让我们开始测试并查看实现前面操作的输出，如下面的屏幕截图所示：

List(1001, 100, 102 ,1)
4
1001
1001
100
102
1

第一个操作返回所有值。第二个操作返回4作为计数。我们将考虑第一个元素1001，但这是一个随机值，它是无序的。然后我们在循环中打印所有的元素，如下面的输出所示：

102
1
1001
1
List(1)
List(100, 1)

然后我们得到max和min值，如1001和1，这与first()类似。之后，我们得到一个有序列表List(1)，和一个样本List(100, 1)，这是随机的。因此，在样本中，我们从输入数据和应用的转换中得到随机值。

在下一节中，我们将学习如何重用rdd进行不同的操作。

重用相同的 rdd 进行不同的操作

在这一部分，我们将重用相同的rdd进行不同的操作。首先，我们将通过重用rdd来最小化执行时间。然后，我们将查看缓存和我们代码的性能测试。

下面的例子是前面部分的测试，但稍作修改，这里我们通过currentTimeMillis()取start和result。因此，我们只是测量执行的所有操作的result：

//then every call to action means that we are going up to the RDD chain
//if we are loading data from external file-system (I.E.: HDFS), every action means
//that we need to load it from FS.
    val start = System.currentTimeMillis()
    println(rdd.collect().toList)
    println(rdd.count())
    println(rdd.first())
    rdd.foreach(println(_))
    rdd.foreachPartition(t => t.foreach(println(_)))
    println(rdd.max())
    println(rdd.min())
    println(rdd.takeOrdered(1).toList)
    println(rdd.takeSample(false, 2).toList)
    val result = System.currentTimeMillis() - start

    println(s"time taken (no-cache): $result")

}

如果有人对 Spark 不太了解，他们会认为所有操作都被巧妙地执行了。我们知道每个操作都意味着我们要上升到链中的rdd，这意味着我们要对所有的转换进行加载数据。在生产系统中，加载数据将来自外部的 PI 系统，比如 HDFS。这意味着每个操作都会导致对文件系统的调用，这将检索所有数据，然后应用转换，如下例所示：

//when apply transformation
val rdd = input
    .filter(_.userId.contains("A"))
    .keyBy(_.userId)
    .map(_._2.amount)

这是一个非常昂贵的操作，因为每个操作都非常昂贵。当我们开始这个测试时，我们可以看到没有缓存的时间为 632 毫秒，如下面的输出所示：

List(1)
List(100, 1)
time taken (no-cache): 632
Process finished with exit code 0

让我们将这与缓存使用进行比较。乍一看，我们的测试看起来非常相似，但这并不相同，因为您正在使用cache()，而我们正在返回rdd。因此，rdd将已经被缓存，对rdd的每个后续调用都将经过cache，如下例所示：

//when apply transformation
val rdd = input
    .filter(_.userId.contains("A"))
    .keyBy(_.userId)
    .map(_._2.amount)
    .cache()

第一个操作将执行 DAG，将数据保存到我们的缓存中，然后后续的操作将根据从内存中调用的方法来检索特定的内容。不会有 HDFS 查找，所以让我们按照以下示例开始这个测试，看看需要多长时间：

//then every call to action means that we are going up to the RDD chain
//if we are loading data from external file-system (I.E.: HDFS), every action means
//that we need to load it from FS.
    val start = System.currentTimeMillis()
    println(rdd.collect().toList)
    println(rdd.count())
    println(rdd.first())
    rdd.foreach(println(_))
    rdd.foreachPartition(t => t.foreach(println(_)))
    println(rdd.max())
    println(rdd.min())
    println(rdd.takeOrdered(1).toList)
    println(rdd.takeSample(false, 2).toList)
    val result = System.currentTimeMillis() - start

    println(s"time taken(cache): $result")

    }
}

第一个输出将如下所示：

List(1)
List(100, 102)
time taken (no-cache): 585
List(1001, 100, 102, 1)
4

第二个输出将如下所示：

1
List(1)
List(102, 1)
time taken(cache): 336
Process finished with exit code 0

没有缓存，值为585毫秒，有缓存时，值为336。这个差异并不大，因为我们只是在测试中创建数据。然而，在真实的生产系统中，这将是一个很大的差异，因为我们需要从外部文件系统中查找数据。

总结

因此，让我们总结一下这一章节。首先，我们使用 Spark 转换来推迟计算到以后的时间，然后我们学习了哪些转换应该避免。接下来，我们看了如何使用reduceByKey和reduce来计算我们的全局结果和特定键的结果。之后，我们执行了触发计算的操作，然后了解到每个操作都意味着加载数据的调用。为了缓解这个问题，我们学习了如何为不同的操作减少相同的rdd。

在下一章中，我们将看一下 Spark 引擎的不可变设计。

第八章：不可变设计

在本章中，我们将看看 Apache Spark 的不可变设计。我们将深入研究 Spark RDD 的父/子链，并以不可变的方式使用 RDD。然后，我们将使用 DataFrame 操作进行转换，以讨论在高度并发的环境中的不可变性。在本章结束时，我们将以不可变的方式使用数据集 API。

在这一章中，我们将涵盖以下主题：

• 深入研究 Spark RDD 的父/子链

• 以不可变的方式使用 RDD

• 使用 DataFrame 操作进行转换

• 在高度并发的环境中的不可变性

• 以不可变的方式使用数据集 API

深入研究 Spark RDD 的父/子链

在本节中，我们将尝试实现我们自己的 RDD，继承 RDD 的父属性。

我们将讨论以下主题：

• 扩展 RDD

• 与父 RDD 链接新的 RDD

• 测试我们的自定义 RDD

扩展 RDD

这是一个有很多隐藏复杂性的简单测试。让我们从创建记录的列表开始，如下面的代码块所示：

class InheritanceRdd extends FunSuite {
  val spark: SparkContext = SparkSession
    .builder().master("local[2]").getOrCreate().sparkContext

  test("use extended RDD") {
    //given
    val rdd = spark.makeRDD(List(Record(1, "d1")))

Record只是一个具有amount和description的案例类，所以amount是 1，d1是描述。

然后我们创建了MultipledRDD并将rdd传递给它，然后将乘数设置为10，如下面的代码所示：

val extendedRdd = new MultipliedRDD(rdd, 10)

我们传递父 RDD，因为它包含在另一个 RDD 中加载的数据。通过这种方式，我们构建了两个 RDD 的继承链。

与父 RDD 链接新的 RDD

我们首先创建了一个多重 RDD 类。在MultipliedRDD类中，我们有两个传递参数的东西：

• 记录的简要 RDD，即RDD[Record]

• 乘数，即Double

在我们的情况下，可能会有多个 RDD 的链，这意味着我们的 RDD 中可能会有多个 RDD。因此，这并不总是所有有向无环图的父级。我们只是扩展了类型为记录的 RDD，因此我们需要传递扩展的 RDD。

RDD 有很多方法，我们可以覆盖任何我们想要的方法。但是，这一次，我们选择了compute方法，我们将覆盖计算乘数的方法。在这里，我们获取Partition分区和TaskContext。这些是执行引擎传递给我们方法的，因此我们不需要担心这一点。但是，我们需要返回与我们通过继承链中的 RDD 类传递的类型完全相同的迭代器。这将是记录的迭代器。

然后我们执行第一个父逻辑，第一个父只是获取我们链中的第一个 RDD。这里的类型是Record，我们获取split和context的iterator，其中split只是将要执行的分区。我们知道 Spark RDD 是由分区器分区的，但是在这里，我们只是获取我们需要拆分的特定分区。因此，迭代器获取分区和任务上下文，因此它知道应该从该迭代方法返回哪些值。对于迭代器中的每条记录，即salesRecord，如amount和description，我们将amount乘以传递给构造函数的multiplier来获得我们的Double。

通过这样做，我们已经将我们的金额乘以了乘数，然后我们可以返回具有新金额的新记录。因此，我们现在有了旧记录乘以我们的“乘数”的金额和salesRecord的描述。对于第二个过滤器，我们需要“覆盖”的是getPartitions，因为我们希望保留父 RDD 的分区。例如，如果之前的 RDD 有 100 个分区，我们也希望我们的MultipledRDD有 100 个分区。因此，我们希望保留关于分区的信息，而不是丢失它。出于同样的原因，我们只是将其代理给firstParent。RDD 的firstParent然后只会从特定 RDD 中获取先前的分区。

通过这种方式，我们创建了一个新的multipliedRDD，它传递了父级和乘数。对于我们的extendedRDD，我们需要collect它并调用toList，我们的列表应该包含10和d1，如下例所示：

extendedRdd.collect().toList should contain theSameElementsAs List(
 Record(10, "d1")
 )
 }
}

当我们创建新的 RDD 时，计算会自动执行，因此它总是在没有显式方法调用的情况下执行。

测试我们的自定义 RDD

让我们开始这个测试，以检查这是否已经创建了我们的 RDD。通过这样做，我们可以扩展我们的父 RDD 并向我们的 RDD 添加行为。这在下面的截图中显示：

"C:\Program Files\Java\jdk-12\bin\java.exe" "-javaagent:C:\Program Files\JetBrains\IntelliJ IDEA 2018.3.5\lib\idea_rt.jar=51687:C:\Program Files\JetBrains\IntelliJ IDEA 2018.3.5\bin" -Dfile.encoding=UTF-8 -classpath C:\Users\Sneha\IdeaProjects\Chapter07\out\production\Chapter07 com.company.Main

Process finished with exit code 0

在下一节中，我们将以不可变的方式使用 RDD。

以不可变的方式使用 RDD

现在我们知道如何使用 RDD 继承创建执行链，让我们学习如何以不可变的方式使用 RDD。

在这一部分，我们将讨论以下主题：

• 理解 DAG 的不可变性

• 从一个根 RDD 创建两个叶子

• 检查两个叶子的结果

让我们首先了解有向无环图的不可变性以及它给我们带来了什么。然后，我们将从一个节点 RDD 创建两个叶子，并检查如果我们在一个叶子 RDD 上创建一个转换，那么两个叶子是否完全独立地行为。然后，我们将检查当前 RDD 的两个叶子的结果，并检查对任何叶子的任何转换是否不会改变或影响根 RDD。以这种方式工作是至关重要的，因为我们发现我们将无法从根 RDD 创建另一个叶子，因为根 RDD 将被更改，这意味着它将是可变的。为了克服这一点，Spark 设计师为我们创建了一个不可变的 RDD。

有一个简单的测试来显示 RDD 应该是不可变的。首先，我们将从0 到 5创建一个 RDD，它被添加到来自 Scala 分支的序列中。to获取Int，第一个参数是一个隐式参数，来自 Scala 包，如下例所示：

class ImmutableRDD extends FunSuite {
    val spark: SparkContext = SparkSession
        .builder().master("local[2]").getOrCreate().sparkContext

test("RDD should be immutable") {
    //given
    val data = spark.makeRDD(0 to 5)

一旦我们有了 RDD 数据，我们可以创建第一个叶子。第一个叶子是一个结果（res），我们只是将每个元素乘以2。让我们创建第二个叶子，但这次它将被标记为4，如下例所示：

//when
val res = data.map(_ * 2)

val leaf2 = data.map(_ * 4)

所以，我们有我们的根 RDD 和两个叶子。首先，我们将收集第一个叶子，并看到其中的元素为0, 2, 4, 6, 8, 10，所以这里的一切都乘以2，如下例所示：

//then
res.collect().toList should contain theSameElementsAs List(
    0, 2, 4, 6, 8, 10
)

然而，即使我们在res上有了通知，数据仍然与一开始的完全相同，即0, 1, 2, 3, 4, 5，如下例所示：

data.collect().toList should contain theSameElementsAs List(
    0, 1, 2, 3, 4, 5
    )
  }
}

所以，一切都是不可变的，执行* 2的转换并没有改变我们的数据。如果我们为leaf2创建一个测试，我们将collect它并调用toList。我们会看到它应该包含像0, 4, 8, 12, 16, 20这样的元素，如下例所示：

leaf2.collect().toList should contain theSameElementsAs List(
 0, 4, 8, 12, 16, 20
)

当我们运行测试时，我们会看到我们执行中的每条路径，即数据或第一个叶子和第二个叶子，彼此独立地行为，如下面的代码输出所示：

"C:\Program Files\Java\jdk-12\bin\java.exe" "-javaagent:C:\Program Files\JetBrains\IntelliJ IDEA 2018.3.5\lib\idea_rt.jar=51704:C:\Program Files\JetBrains\IntelliJ IDEA 2018.3.5\bin" -Dfile.encoding=UTF-8 -classpath C:\Users\Sneha\IdeaProjects\Chapter07\out\production\Chapter07 com.company.Main

Process finished with exit code 0

每次变异都是不同的；我们可以看到测试通过了，这表明我们的 RDD 是不可变的。

使用 DataFrame 操作进行转换

API 的数据下面有一个 RDD，因此 DataFrame 是不可变的。在 DataFrame 中，不可变性甚至更好，因为我们可以动态地添加和减去列，而不改变源数据集。

在这一部分，我们将涵盖以下主题：

• 理解 DataFrame 的不可变性

• 从一个根 DataFrame 创建两个叶子

• 通过发出转换来添加新列

我们将首先使用操作的数据来转换我们的 DataFrame。首先，我们需要了解 DataFrame 的不可变性，然后我们将从一个根 DataFrame 创建两个叶子，但这次是。然后，我们将发出一个略有不同于 RDD 的转换。这将向我们的结果 DataFrame 添加一个新列，因为我们在 DataFrame 中是这样操作的。如果我们想要映射数据，那么我们需要从第一列中获取数据，进行转换，并保存到另一列，然后我们将有两列。如果我们不再感兴趣，我们可以删除第一列，但结果将是另一个 DataFrame。

因此，我们将有第一个 DataFrame 有一列，第二个有结果和源，第三个只有一个结果。让我们看看这一部分的代码。

我们将创建一个 DataFrame，所以我们需要调用toDF()方法。我们将使用"a"作为"1"，"b"作为"2"，"d"作为"200"来创建UserData。UserData有userID和data两个字段，都是String类型，如下例所示：

test("Should use immutable DF API") {
 import spark.sqlContext.implicits._
 //given
 val userData =
 spark.sparkContext.makeRDD(List(
 UserData("a", "1"),
 UserData("b", "2"),
 UserData("d", "200")
 )).toDF()

在测试中使用案例类创建 RDD 是很重要的，因为当我们调用 DataFrame 时，这部分将推断模式并相应地命名列。以下代码是这方面的一个例子，我们只从userData中的userID列中进行过滤：

//when
    val res = userData.filter(userData("userId").isin("a"))

我们的结果应该只有一条记录，所以我们要删除两列，但是我们创建的userData源将有 3 行。因此，通过过滤对其进行修改，创建了另一个名为res的 DataFrame，而不修改输入的userData，如下例所示：

    assert(res.count() == 1)
    assert(userData.count() == 3)

    }
}

让我们开始这个测试，看看来自 API 的不可变数据的行为，如下屏幕截图所示：

"C:\Program Files\Java\jdk-12\bin\java.exe" "-javaagent:C:\Program Files\JetBrains\IntelliJ IDEA 2018.3.5\lib\idea_rt.jar=51713:C:\Program Files\JetBrains\IntelliJ IDEA 2018.3.5\bin" -Dfile.encoding=UTF-8 -classpath C:\Users\Sneha\IdeaProjects\Chapter07\out\production\Chapter07 com.company.Main

Process finished with exit code 0

正如我们所看到的，我们的测试通过了，并且从结果（res）中，我们知道我们的父级没有被修改。因此，例如，如果我们想在res.map()上映射一些东西，我们可以映射userData列，如下例所示：

res.map(a => a.getString("userId") + "can")

另一个叶子将具有一个额外的列，而不更改userId源代码，因此这就是 DataFrame 的不可变性。

高并发环境中的不可变性

我们看到了不可变性如何影响程序的创建和设计，现在我们将了解它的用途。

在本节中，我们将涵盖以下主题：

• 可变集合的缺点

• 创建两个同时修改可变集合的线程

• 推理并发程序

让我们首先了解可变集合的原因。为此，我们将创建两个同时修改可变集合的线程。我们将使用此代码进行测试。首先，我们将创建一个ListBuffer，它是一个可变列表。然后，我们可以添加和删除链接，而无需为任何修改创建另一个列表。然后，我们可以创建一个具有两个线程的Executors服务。我们需要两个线程同时开始修改状态。稍后，我们将使用Java.util.concurrent中的CountDownLatch构造。这在下面的例子中显示：

import java.util.concurrent.{CountDownLatch, Executors}
import org.scalatest.FunSuite
import scala.collection.mutable.ListBuffer
class MultithreadedImmutabilityTest extends FunSuite {

test("warning: race condition with mutability") {
//given
var listMutable = new ListBuffer[String]()
val executors = Executors.newFixedThreadPool(2)
val latch = new CountDownLatch(2)

CountDownLatch是一种构造，它帮助我们阻止线程处理，直到我们要求它们开始。我们需要等待逻辑，直到两个线程开始执行。然后，我们向executors提交一个Runnable，我们的run()方法通过发出countDown()来表示准备好进行操作，并将"A"添加到listMutable，如下例所示：

 //when
 executors.submit(new Runnable {
     override def run(): Unit = {
         latch.countDown()
         listMutable += "A"
     }
 })

然后，另一个线程启动，并且也使用countDown来表示它已准备好开始。但首先，它会检查列表是否包含"A"，如果没有，就会添加"A"，如下例所示：

 executors.submit(new Runnable {
     override def run(): Unit = {
         latch.countDown()
         if(!listMutable.contains("A")) {
             listMutable += "A"
         }
     }
 })

然后，我们使用await()等待countDown发出，当它发出时，我们可以继续验证我们的程序，如下例所示：

    latch.await()

listMutable包含"A"或可能包含"A","A"。listMutable检查列表是否包含"A",如果没有，它将不会添加该元素，如下例所示：

    //then
    //listMutable can have ("A") or ("A","A")
    }
}

但这里存在竞争条件。在检查if(!listMutable.contains("A"))之后，run()线程可能会将"A"元素添加到列表中。但我们在if中，所以我们将通过listMutable += "A"添加另一个"A"。由于状态的可变性以及它通过另一个线程进行了修改，我们可能会有"A"或"A","A"。

在使用可变状态时需要小心，因为我们不能有这样一个损坏的状态。为了缓解这个问题，我们可以在java.util集合上使用同步列表。

但如果我们有同步块，那么我们的程序将非常慢，因为我们需要独占地访问它。我们还可以使用java.util.concurrent.locks包中的lock。我们可以使用ReadLock或WriteLock等实现。在下面的例子中，我们将使用WriteLock：

val lock = new WriteLock()

我们还需要对我们的lock()进行lock，然后再进行下一步，如下例所示：

lock.lock()

之后，我们可以使用unlock()。然而，我们也应该在第二个线程中这样做，这样我们的列表只有一个元素，如下例所示：

lock.unlock()

输出如下：

锁定是一个非常艰难和昂贵的操作，因此不可变性是性能程序的关键。

以不可变的方式使用数据集 API

在本节中，我们将以不可变的方式使用数据集 API。我们将涵盖以下主题：

• 数据集的不可变性

• 从一个根数据集创建两个叶子

• 通过发出转换添加新列

数据集的测试用例非常相似，但我们需要对我们的数据进行toDS()以确保类型安全。数据集的类型是userData，如下例所示：

import com.tomekl007.UserData
import org.apache.spark.sql.SparkSession
import org.scalatest.FunSuite

class ImmutableDataSet extends FunSuite {
 val spark: SparkSession = SparkSession
 .builder().master("local[2]").getOrCreate()

test("Should use immutable DF API") {
 import spark.sqlContext.implicits._
 //given
 val userData =
 spark.sparkContext.makeRDD(List(
 UserData("a", "1"),
 UserData("b", "2"),
 UserData("d", "200")
 )).toDF()

现在，我们将发出对userData的过滤，并指定isin，如下例所示：

   //when
    val res = userData.filter(userData("userId").isin("a"))

它将返回结果（res），这是一个带有我们的1元素的叶子。由于这个明显的根，userData仍然有3个元素。让我们执行这个程序，如下例所示：

    assert(res.count() == 1)
    assert(userData.count() == 3)

 }
}

我们可以看到我们的测试通过了，这意味着数据集也是 DataFrame 之上的不可变抽象，并且具有相同的特性。userData有一个非常有用的类型集，如果使用show()方法，它将推断模式并知道"a"字段是字符串或其他类型，如下例所示：

userData.show()

输出将如下所示：

+------+----+
|userId|data|
|----- |----|
|     a|   1|
|     b|   2|
|     d| 200|
+------|----+ 

在前面的输出中，我们有userID和data字段。

总结

在本章中，我们深入研究了 Spark RDD 的父子链，并创建了一个能够根据父 RDD 计算一切的乘数 RDD，还基于父 RDD 的分区方案。我们以不可变的方式使用了 RDD。我们看到，从父级创建的叶子的修改并没有修改部分。我们还学习了一个更好的抽象，即 DataFrame，因此我们学会了可以在那里使用转换。然而，每个转换只是添加到另一列，而不是直接修改任何内容。接下来，我们只需在高度并发的环境中设置不可变性。我们看到了当访问多个线程时，可变状态是不好的。最后，我们看到数据集 API 也是以不可变的方式创建的，我们可以在这里利用这些特性。

在下一章中，我们将看看如何避免洗牌和减少个人开支。

第九章：避免洗牌和减少操作费用

在本章中，我们将学习如何避免洗牌并减少我们作业的操作费用，以及检测过程中的洗牌。然后，我们将测试在 Apache Spark 中导致洗牌的操作，以找出我们何时应该非常小心以及我们应该避免哪些操作。接下来，我们将学习如何改变具有广泛依赖关系的作业设计。之后，我们将使用keyBy()操作来减少洗牌，在本章的最后一节中，我们将看到如何使用自定义分区来减少数据的洗牌。

在本章中，我们将涵盖以下主题：

• 检测过程中的洗牌

• 在 Apache Spark 中进行导致洗牌的测试操作

• 改变具有广泛依赖关系的作业设计

• 使用keyBy()操作来减少洗牌

• 使用自定义分区器来减少洗牌

检测过程中的洗牌

在本节中，我们将学习如何检测过程中的洗牌。

在本节中，我们将涵盖以下主题：

• 加载随机分区的数据

• 使用有意义的分区键发出重新分区

• 通过解释查询来理解洗牌是如何发生的

我们将加载随机分区的数据，以查看数据是如何加载的以及数据加载到了哪里。接下来，我们将使用有意义的分区键发出一个分区。然后，我们将使用确定性和有意义的键将数据重新分区到适当的执行程序。最后，我们将使用explain()方法解释我们的查询并理解洗牌。在这里，我们有一个非常简单的测试。

我们将创建一个带有一些数据的 DataFrame。例如，我们创建了一个带有一些随机 UID 和user_1的InputRecord，以及另一个带有user_1中随机 ID 的输入，以及user_2的最后一条记录。假设这些数据是通过外部数据系统加载的。数据可以从 HDFS 加载，也可以从数据库加载，例如 Cassandra 或 NoSQL：

class DetectingShuffle extends FunSuite {
  val spark: SparkSession = SparkSession.builder().master("local[2]").getOrCreate()

  test("should explain plan showing logical and physical with UDF and DF") {
    //given
    import spark.sqlContext.implicits._
    val df = spark.sparkContext.makeRDD(List(
      InputRecord("1234-3456-1235-1234", "user_1"),
      InputRecord("1123-3456-1235-1234", "user_1"),
      InputRecord("1123-3456-1235-9999", "user_2")
    )).toDF()

在加载的数据中，我们的数据没有预定义或有意义的分区，这意味着输入记录编号 1 可能会最先出现在执行程序中，而记录编号 2 可能会最先出现在执行程序中。因此，即使数据来自同一用户，我们也很可能会为特定用户执行操作。

如前一章第八章中所讨论的，不可变设计，我们使用了reducebyKey()方法，该方法获取用户 ID 或特定 ID 以减少特定键的所有值。这是一个非常常见的操作，但具有一些随机分区。最好使用有意义的键repartition数据。

在使用userID时，我们将使用repartition的方式，使结果记录具有相同用户 ID 的数据。因此，例如user_1最终将出现在第一个执行程序上：

//when
    val q = df.repartition(df("userId"))

第一个执行程序将拥有所有userID的数据。如果InputRecord("1234-3456-1235-1234", "user_1")在执行程序 1 上，而InputRecord("1123-3456-1235-1234", "user_1")在执行程序 2 上，在对来自执行程序 2 的数据进行分区后，我们需要将其发送到执行程序 1，因为它是此分区键的父级。这会导致洗牌。洗牌是由于加载数据而导致的，这些数据没有有意义地分区，或者根本没有分区。我们需要处理我们的数据，以便我们可以对特定键执行操作。

我们可以进一步repartition数据，但应该在链的开头进行。让我们开始测试来解释我们的查询：

 q.explain(true)

我们在逻辑计划中对userID表达式进行了重新分区，但当我们检查物理计划时，显示使用了哈希分区，并且我们将对userID值进行哈希处理。因此，我们扫描所有 RDD 和所有具有相同哈希的键，并将其发送到相同的执行程序以实现我们的目标：

！

在下一节中，我们将测试在 Apache Spark 中导致洗牌的操作。

在 Apache Spark 中进行导致洗牌的测试操作

在本节中，我们将测试在 Apache Spark 中导致洗牌的操作。我们将涵盖以下主题：

• 使用 join 连接两个 DataFrame

• 使用分区不同的两个 DataFrame

• 测试导致洗牌的连接

连接是一种特定的操作，会导致洗牌，我们将使用它来连接我们的两个 DataFrame。我们将首先检查它是否会导致洗牌，然后我们将检查如何避免它。为了理解这一点，我们将使用两个分区不同的 DataFrame，并检查连接两个未分区或随机分区的数据集或 DataFrame 的操作。如果它们位于不同的物理机器上，将会导致洗牌，因为没有办法连接具有相同分区键的两个数据集。

在我们连接数据集之前，我们需要将它们发送到同一台物理机器上。我们将使用以下测试。

我们需要创建UserData，这是一个我们已经见过的案例类。它有用户 ID 和数据。我们有用户 ID，即user_1，user_2和user_4：

test("example of operation that is causing shuffle") {
    import spark.sqlContext.implicits._
    val userData =
    spark.sparkContext.makeRDD(List(
        UserData("user_1", "1"),
        UserData("user_2", "2"),
        UserData("user_4", "200")
    )).toDS()

然后我们创建一些类似于用户 ID（user_1，user_2和user_3）的交易数据：

val transactionData =
    spark.sparkContext.makeRDD(List(
        UserTransaction("user_1", 100),
        UserTransaction("user_2", 300),
        UserTransaction("user_3", 1300)
    )).toDS()

我们使用joinWith在UserData上的交易，使用UserData和transactionData的userID列。由于我们发出了inner连接，结果有两个元素，因为记录和交易之间有连接，即UserData和UserTransaction。但是，UserData没有交易，Usertransaction没有用户数据：

//shuffle: userData can stay on the current executors, but data from
//transactionData needs to be send to those executors according to joinColumn
//causing shuffle
//when
val res: Dataset[(UserData, UserTransaction)]
= userData.joinWith(transactionData, userData("userId") === transactionData("userId"), "inner")

当我们连接数据时，数据没有分区，因为这是 Spark 的一些随机数据。它无法知道用户 ID 列是分区键，因为它无法猜测。由于它没有预分区，要连接来自两个数据集的数据，需要将数据从用户 ID 发送到执行器。因此，由于数据没有分区，将会有大量数据从执行器洗牌。

让我们解释查询，执行断言，并通过启动测试显示结果：

//then
 res.show()
 assert(res.count() == 2)
 }
}

我们可以看到我们的结果如下：

+------------+-------------+
|         _1 |           _2|
+----------- +-------------+
+ [user_1,1] | [user_1,100]|
| [user_2,2] | [user_2,300]|
+------------+-------------+

我们有[user_1,1]和[user_1,100]，即userID和userTransaction。看起来连接工作正常，但让我们看看物理参数。我们使用SortMergeJoin对第一个数据集和第二个数据集使用userID，然后我们使用Sort和hashPartitioning。

在前一节中，检测过程中的洗牌，我们使用了partition方法，该方法在底层使用了hashPartitioning。虽然我们使用了join，但我们仍然需要使用哈希分区，因为我们的数据没有正确分区。因此，我们需要对第一个数据集进行分区，因为会有大量的洗牌，然后我们需要对第二个 DataFrame 做完全相同的事情。再次，洗牌将会进行两次，一旦数据根据连接字段进行分区，连接就可以在执行器本地进行。

在执行物理计划后，将对记录进行断言，指出userID用户数据一与用户交易userID一位于同一执行器上。没有hashPartitioning，就没有保证，因此我们需要进行分区。

在下一节中，我们将学习如何更改具有广泛依赖的作业的设计，因此我们将看到如何在连接两个数据集时避免不必要的洗牌。

更改具有广泛依赖的作业的设计

在本节中，我们将更改在未分区数据上执行join的作业。我们将更改具有广泛依赖的作业的设计。

在本节中，我们将涵盖以下主题：

• 使用公共分区键对 DataFrame 进行重新分区

• 理解使用预分区数据进行连接

• 理解我们如何避免洗牌

我们将在 DataFrame 上使用repartition方法，使用一个公共分区键。我们发现，当进行连接时，重新分区会在底层发生。但通常，在使用 Spark 时，我们希望在 DataFrame 上执行多个操作。因此，当我们与其他数据集执行连接时，hashPartitioning将需要再次执行。如果我们在加载数据时进行分区，我们将避免再次分区。

在这里，我们有我们的示例测试用例，其中包含我们之前在 Apache Spark 的“导致洗牌的测试操作”部分中使用的数据。我们有UserData，其中包含三条用户 ID 的记录 - user_1，user_2和user_4 - 以及UserTransaction数据，其中包含用户 ID - 即user_1，user_2，user_3：

test("example of operation that is causing shuffle") {
    import spark.sqlContext.implicits._
    val userData =
        spark.sparkContext.makeRDD(List(
            UserData("user_1", "1"),
            UserData("user_2", "2"),
            UserData("user_4", "200")
        )).toDS()

然后，我们需要对数据进行repartition，这是要做的第一件非常重要的事情。我们使用userId列来重新分区我们的userData：

val repartitionedUserData = userData.repartition(userData("userId"))

然后，我们将使用userId列重新分区我们的数据，这次是针对transactionData：

 val repartitionedTransactionData = transactionData.repartition(transactionData("userId"))

一旦我们重新分区了我们的数据，我们就可以确保具有相同分区键的任何数据 - 在本例中是userId - 将落在同一个执行器上。因此，我们的重新分区数据将不会有洗牌，连接将更快。最终，我们能够进行连接，但这次我们连接的是预分区的数据：

//when
//data is already partitioned using join-column. Don't need to shuffle
val res: Dataset[(UserData, UserTransaction)]
= repartitionedUserData.joinWith(repartitionedTransactionData, userData("userId") === transactionData("userId"), "inner")

我们可以使用以下代码显示我们的结果：

 //then
 res.show()
 assert(res.count() == 2)
 }
}

输出显示在以下截图中：

在上述截图中，我们有用户 ID 和交易的物理计划。我们对用户 ID 数据和交易数据的用户 ID 列执行了哈希分区。在连接数据之后，我们可以看到数据是正确的，并且连接有一个物理计划。

这次，物理计划有点不同。

我们有一个SortMergeJoin操作，并且我们正在对我们的数据进行排序，这些数据在我们执行引擎的上一步已经预分区。这样，我们的 Spark 引擎将执行排序合并连接，无需进行哈希连接。它将正确排序数据，连接将更快。

在下一节中，我们将使用keyBy()操作来进一步减少洗牌。

使用 keyBy()操作来减少洗牌

在本节中，我们将使用keyBy()操作来减少洗牌。我们将涵盖以下主题：

• 加载随机分区的数据

• 尝试以有意义的方式预分区数据

• 利用keyBy()函数

我们将加载随机分区的数据，但这次使用 RDD API。我们将以有意义的方式重新分区数据，并提取底层正在进行的信息，类似于 DataFrame 和 Dataset API。我们将学习如何利用keyBy()函数为我们的数据提供一些结构，并在 RDD API 中引起预分区。

本节中我们将使用以下测试。我们创建两个随机输入记录。第一条记录有一个随机用户 ID，user_1，第二条记录有一个随机用户 ID，user_1，第三条记录有一个随机用户 ID，user_2：

test("Should use keyBy to distribute traffic properly"){
    //given
    val rdd = spark.sparkContext.makeRDD(List(
        InputRecord("1234-3456-1235-1234", "user_1"),
        InputRecord("1123-3456-1235-1234", "user_1"),
        InputRecord("1123-3456-1235-9999", "user_2")
    ))

我们将使用rdd.toDebugString提取 Spark 底层发生的情况：

println(rdd.toDebugString)

此时，我们的数据是随机分布的，用户 ID 字段的记录可能在不同的执行器上，因为 Spark 执行引擎无法猜测user_1是否对我们有意义，或者1234-3456-1235-1234是否有意义。我们知道1234-3456-1235-1234不是一个有意义的键，而是一个唯一标识符。将该字段用作分区键将给我们一个随机分布和大量的洗牌，因为在使用唯一字段作为分区键时没有数据局部性。

Spark 无法知道相同用户 ID 的数据将落在同一个执行器上，这就是为什么在分区数据时我们需要使用用户 ID 字段，即user_1、user_1或user_2。为了在 RDD API 中实现这一点，我们可以在我们的数据中使用keyBy(_.userId)，但这次它将改变 RDD 类型：

val res = rdd.keyBy(_.userId)

如果我们检查 RDD 类型，我们会发现这次，RDD 不是输入记录，而是字符串和输入记录的 RDD。字符串是我们在这里期望的字段类型，即userId。我们还将通过在结果上使用toDebugString来提取有关keyBy()函数的信息：

println(res.toDebugString)

一旦我们使用keyBy()，相同用户 ID 的所有记录都将落在同一个执行器上。正如我们所讨论的，这可能是危险的，因为如果我们有一个倾斜的键，这意味着我们有一个具有非常高基数的键，我们可能会耗尽内存。此外，结果上的所有操作都将按键进行，因此我们将在预分区数据上进行操作：

res.collect()

让我们开始这个测试。输出将如下所示：

我们可以看到我们的第一个调试字符串非常简单，我们只有 RDD 上的集合，但第二个有点不同。我们有一个keyBy()方法，并在其下面创建了一个 RDD。我们有来自第一部分的子 RDD 和父 RDD，即测试在 Apache Spark 中引起洗牌的操作，当我们扩展了 RDD 时。这是由keyBy()方法发出的父子链。

在下一节中，我们将使用自定义分区器进一步减少洗牌。

使用自定义分区器来减少洗牌

在本节中，我们将使用自定义分区器来减少洗牌。我们将涵盖以下主题：

• 实现自定义分区器

• 使用partitionBy方法在 Spark 上使用分区器

• 验证我们的数据是否被正确分区

我们将使用自定义逻辑实现自定义分区器，该分区器将对数据进行分区。它将告诉 Spark 每条记录应该落在哪个执行器上。我们将使用 Spark 上的partitionBy方法。最后，我们将验证我们的数据是否被正确分区。为了测试的目的，我们假设有两个执行器：

import com.tomekl007.UserTransaction
import org.apache.spark.sql.SparkSession
import org.apache.spark.{Partitioner, SparkContext}
import org.scalatest.FunSuite
import org.scalatest.Matchers._

class CustomPartitioner extends FunSuite {
val spark: SparkContext = SparkSession.builder().master("local[2]").getOrCreate().sparkContext

test("should use custom partitioner") {
//given
val numberOfExecutors = 2

假设我们想将我们的数据均匀地分成2个执行器，并且具有相同键的数据实例将落在同一个执行器上。因此，我们的输入数据是一个UserTransactions列表："a","b","a","b"和"c"。值并不那么重要，但我们需要记住它们以便稍后测试行为。给定UserTransactions的amount分别为100,101,202,1和55：

val data = spark
    .parallelize(List(
        UserTransaction("a", 100),
        UserTransaction("b", 101),
        UserTransaction("a", 202),
        UserTransaction("b", 1),
        UserTransaction("c", 55)

当我们使用keyBy时，(_.userId)被传递给我们的分区器，因此当我们发出partitionBy时，我们需要扩展override方法：

).keyBy(_.userId)
.partitionBy(new Partitioner {
    override def numPartitions: Int = numberOfExecutors

getPartition方法接受一个key，它将是userId。键将在这里传递，类型将是字符串：

override def getPartition(key: Any): Int = {
    key.hashCode % numberOfExecutors
    }
})

这些方法的签名是Any，所以我们需要override它，并且还需要覆盖分区的数量。

然后我们打印我们的两个分区，numPartitions返回值为2：

println(data.partitions.length)

getPartition非常简单，因为它获取hashCode和numberOfExecutors的模块。它确保相同的键将落在同一个执行器上。

然后，我们将为各自的分区映射每个分区，因为我们得到一个迭代器。在这里，我们正在为测试目的获取amount：

//when
val res = data.mapPartitionsLong.map(_.amount)
).collect().toList

最后，我们断言55,100,202,101和1；顺序是随机的，所以不需要关心顺序：

//then
res should contain theSameElementsAs List(55, 100, 202, 101, 1)
}
}

如果我们仍然希望，我们应该使用sortBy方法。让我们开始这个测试，看看我们的自定义分区器是否按预期工作。现在，我们可以开始了。我们有2个分区，所以它按预期工作，如下面的截图所示：

总结

在本章中，我们学习了如何检测过程中的洗牌。我们涵盖了在 Apache Spark 中导致洗牌的测试操作。我们还学习了如何在 RDD 中使用分区。如果需要分区数据，了解如何使用 API 是很重要的，因为 RDD 仍然被广泛使用，所以我们使用keyBy操作来减少洗牌。我们还学习了如何使用自定义分区器来减少洗牌。

在下一章中，我们将学习如何使用 Spark API 以正确的格式保存数据。

第十章：将数据保存在正确的格式中

在之前的章节中，我们专注于处理和加载数据。我们学习了有关转换、操作、连接、洗牌和 Spark 的其他方面。

在本章中，我们将学习如何以正确的格式保存数据，还将使用 Spark 的标准 API 以纯文本格式保存数据。我们还将利用 JSON 作为数据格式，并学习如何使用标准 API 保存 JSON。Spark 有 CSV 格式，我们也将利用该格式。然后，我们将学习更高级的基于模式的格式，其中需要支持导入第三方依赖项。接下来，我们将使用 Avro 与 Spark，并学习如何使用和保存列格式的数据，即 Parquet。到本章结束时，我们还将学会如何检索数据以验证其是否以正确的方式存储。

在本章中，我们将涵盖以下主题：

• 以纯文本格式保存数据

• 利用 JSON 作为数据格式

• 表格式 - CSV

• 使用 Avro 与 Spark

• 列格式 - Parquet

以纯文本格式保存数据

在本节中，我们将学习如何以纯文本格式保存数据。将涵盖以下主题：

• 以纯文本格式保存数据

• 加载纯文本数据

• 测试

我们将以纯文本格式保存我们的数据，并研究如何将其保存到 Spark 目录中。然后我们将加载纯文本数据，然后测试并保存以检查我们是否可以产生相同的结果代码。这是我们的SavePlainText.scala文件：

package com.tomekl007.chapter_4

import java.io.File

import com.tomekl007.UserTransaction
import org.apache.spark.sql.SparkSession
import org.apache.spark.{Partitioner, SparkContext}
import org.scalatest.{BeforeAndAfterEach, FunSuite}
import org.scalatest.Matchers._

import scala.reflect.io.Path

class SavePlainText extends FunSuite with BeforeAndAfterEach{
    val spark: SparkContext = SparkSession.builder().master("local[2]").getOrCreate().sparkContext

    private val FileName = "transactions.txt"

    override def afterEach() {
        val path = Path (FileName)
        path.deleteRecursively()
    }

    test("should save and load in plain text") {
        //given
        val rdd = spark.makeRDD(List(UserTransaction("a", 100), UserTransaction("b", 200)))

        //when
        rdd.coalesce(1).saveAsTextFile(FileName)

        val fromFile = spark.textFile(FileName)

        fromFile.collect().toList should contain theSameElementsAs List(
            "UserTransaction(a,100)", "UserTransaction(b,200)"
            //note - this is string!
        )
    }
}

我们将需要一个FileName变量，在我们的情况下，它将是一个文件夹名称，然后 Spark 将在其下创建一些文件：

import java.io.File
import com.tomekl007.UserTransaction
import org.apache.spark.sql.SparkSession
import org.apache.spark.{Partitioner, SparkContext}
import org.scalatest.{BeforeAndAfterEach, FunSuite}
import org.scalatest.Matchers._
import scala.reflect.io.Path
class SavePlainText extends FunSuite with BeforeAndAfterEach{
    val spark: SparkContext = SparkSession.builder().master("local[2]").getOrCreate().sparkContext
    private val FileName = "transactions.txt"

我们将在我们的测试用例中使用BeforeAndAfterEach来清理我们的目录，这意味着路径应该被递归删除。测试后整个路径将被删除，因为需要重新运行测试而没有失败。我们需要注释掉以下代码，以便在第一次运行时调查保存的文本文件的结构：

//override def afterEach() {
//         val path = Path (FileName)
//         path.deleteRecursively()
//     }

//test("should save and load in plain text") {

然后我们将创建两个交易的 RDD，UserTransaction("a", 100)和UserTransaction("b", 200)：

val rdd = spark.makeRDD(List(UserTransaction("a", 100), UserTransaction("b", 200)))

然后，我们将我们的数据合并为一个分区。coalesce()是一个非常重要的方面。如果我们想将数据保存在单个文件中，我们需要将其合并为一个，但这样做有一个重要的含义：

rdd.coalesce(1).saveAsTextFile(FileName)

如果我们将其合并为一个文件，那么只有一个执行程序可以将数据保存到我们的系统中。这意味着保存数据将非常缓慢，并且还存在内存不足的风险，因为所有数据将被发送到一个执行程序。通常，在生产环境中，我们根据可用的执行程序将其保存为多个分区，甚至乘以自己的因子。因此，如果我们有 16 个执行程序，那么我们可以将其保存为64。但这会导致64个文件。出于测试目的，我们将保存为一个文件，如前面的代码片段所示：

rdd.coalesce (numPartitions = 1).saveAsTextFile(FileName)

现在，我们将加载数据。我们只需要将文件名传递给TextFile方法，它将返回fromFile：

    val fromFile = spark.textFile(FileName)

然后我们断言我们的数据，这将产生theSameElementsAS List，UserTransaction(a,100)和UserTransaction(b,200)：

    fromFile.collect().toList should contain theSameElementsAs List(
      "UserTransaction(a,100)", "UserTransaction(b,200)"
      //note - this is string!
    )
  }
}

需要注意的重要事项是，对于字符串列表，Spark 不知道我们的数据模式，因为我们将其保存为纯文本。

这是在保存纯文本时需要注意的一点，因为加载数据并不容易，因为我们需要手动将每个字符串映射到UserTransaction。因此，我们将不得不手动解析每条记录，但是，出于测试目的，我们将把我们的交易视为字符串。

现在，让我们开始测试并查看创建的文件夹的结构：

在前面的屏幕截图中，我们可以看到我们的测试通过了，我们得到了transactions.txt。在文件夹中，我们有四个文件。第一个是._SUCCESS.crc，这意味着保存成功。接下来，我们有.part-00000.crc，用于控制和验证一切是否正常工作，这意味着保存是正确的。然后，我们有_SUCCESS和part-00000，这两个文件都有校验和，但part-00000也包含了所有的数据。然后，我们还有UserTransaction(a,100)和UserTransaction(b,200)：

在下一节中，我们将学习如果增加分区数量会发生什么。

利用 JSON 作为数据格式

在本节中，我们将利用 JSON 作为数据格式，并将我们的数据保存为 JSON。以下主题将被涵盖：

• 以 JSON 格式保存数据

• 加载 JSON 数据

• 测试

这些数据是人类可读的，并且比简单的纯文本给我们更多的含义，因为它携带了一些模式信息，比如字段名。然后，我们将学习如何以 JSON 格式保存数据并加载我们的 JSON 数据。

我们将首先创建一个UserTransaction("a", 100)和UserTransaction("b", 200)的 DataFrame，并使用.toDF()保存 DataFrame API：

val rdd = spark.sparkContext
         .makeRDD(List(UserTransaction("a", 100), UserTransaction("b", 200)))
         .toDF()

然后我们将发出coalesce()，这次我们将取值为2，并且我们将得到两个结果文件。然后我们将发出write.format方法，并且需要指定一个格式，我们将使用json格式：

rdd.coalesce(2).write.format("json").save(FileName)

如果我们使用不支持的格式，我们将得到一个异常。让我们通过将源输入为not来测试这一点：

rdd.coalesce(2).write.format("not").save(FileName)

我们将得到诸如“此格式不是预期的”、“找不到数据源：not”和“没有这样的数据源”等异常：

在我们原始的 JSON 代码中，我们将指定格式，并且需要将其保存到FileName。如果我们想要读取，我们需要将其指定为read模式，并且还需要添加一个文件夹的路径：

val fromFile = spark.read.json(FileName)

在这种情况下，让我们注释掉afterEach()来调查生成的 JSON：

// override def afterEach() {
// val path = Path(FileName)
// path.deleteRecursively()
// }

让我们开始测试：

 fromFile.show()
 assert(fromFile.count() == 2)
 }
}

输出如下：

+------+------+
|amount|userId|
|   200|     b|
|   100|     a|
+------+------+

在前面的代码输出中，我们可以看到我们的测试通过了，并且 DataFrame 包含了所有有意义的数据。

从输出中，我们可以看到 DataFrame 具有所需的所有模式。它有amount和userId，这非常有用。

transactions.json文件夹有两部分——一部分是r-00000，另一部分是r-00001，因为我们发出了两个分区。如果我们在生产系统中保存数据有 100 个分区，我们最终会得到 100 个部分文件，而且每个部分文件都会有一个 CRC 校验和文件。

这是第一个文件：

{"userId":"a","amount":"100"}

在这里，我们有一个带有模式的 JSON 文件，因此我们有一个userID字段和amount字段。

另一方面，我们有第二个文件，其中包含第二条记录，包括userID和amount：

{"userId":"b","amount":"200"}

这样做的好处是 Spark 能够从模式中推断出数据，并且以格式化的 DataFrame 加载，具有适当的命名和类型。然而，缺点是每条记录都有一些额外的开销。每条记录都需要在其中有一个字符串，并且在每个字符串中，如果我们有一个包含数百万个文件的文件，并且我们没有对其进行压缩，那么将会有相当大的开销，这是不理想的。

JSON 是人类可读的，但另一方面，它消耗了大量资源，就像 CPU 用于压缩、读取和写入，以及磁盘和内存用于开销一样。除了 JSON 之外，还有更好的格式，我们将在接下来的部分中介绍。

在下一节中，我们将查看表格格式，我们将介绍一个经常用于导入到 Microsoft Excel 或 Google 电子表格的 CSV 文件。这对数据科学家也是非常有用的格式，但仅在使用较小的数据集时。

表格式——CSV

在本节中，我们将介绍文本数据，但以表格格式——CSV。以下主题将被涵盖：

• 以 CSV 格式保存数据

• 加载 CSV 数据

• 测试

保存 CSV 文件比 JSON 和纯文本更复杂，因为我们需要指定是否要在 CSV 文件中保留数据的头信息。

首先，我们将创建一个 DataFrame：

test("should save and load CSV with header") {
 //given
 import spark.sqlContext.implicits._
 val rdd = spark.sparkContext
 .makeRDD(List(UserTransaction("a", 100), UserTransaction("b", 200)))
 .toDF()

然后，我们将使用write格式 CSV。我们还需要指定我们不想在其中包含header选项：

//when
rdd.coalesce(1)
    .write
    .format("csv")
    .option("header", "false")
    .save(FileName)

然后，我们将进行测试以验证条件是true还是false：

    //when
    rdd.coalesce(1)
      .write
      .format("csv")
      .option("header", "true")
      .save(FileName)  

此外，我们无需添加任何额外的依赖来支持 CSV，如以前的版本所需。

然后，我们将指定应该与write模式相似的read模式，并且我们需要指定是否有header：

val fromFile = spark.read.option("header", "false").csv(FileName)

让我们开始测试并检查输出：

+---+---+
|_c0|_c1|
+---+---+
|  a|100|
|  b|200|
+---+---+

在前面的代码输出中，我们可以看到数据已加载，但我们丢失了我们的模式。c0和c1是由 Spark 创建的列 0（c0）和列 1（c1）的别名。

因此，如果我们指定header应保留该信息，让我们在write和read时指定header：

val fromFile = spark.read.option("header", "true).csv(FileName)

我们将指定header应保留我们的信息。在以下输出中，我们可以看到关于模式的信息在读写操作中被感知到：

+------+------+
|userId|amount|
+------+------+
|     a|   100|
|     b|   200|
+------+------+

让我们看看如果我们在write时使用header，而在read时不使用它会发生什么。我们的测试应该失败，如下面的代码截图所示：

在前面的截图中，我们可以看到我们的测试失败了，因为我们没有模式，因为我们在没有头的情况下进行读取。第一条记录，也就是header，被视为列值。

让我们尝试一个不同的情况，我们在没有header的情况下进行写入，并在有header的情况下进行读取：

  //when
 rdd.coalesce(1)
     .write
     .format("csv")
     .option("header", "false")
     .save(FileName)

val fromFile = spark.read.option("header", "false").csv(FileName)

我们的测试将再次失败，因为这一次，我们将我们的第一条记录视为头记录。

让我们将读和写操作都设置为header并在之前添加的注释后测试我们的代码：

override def afterEach() {
    val path = Path(FileName)
    path.deleteRecursively()
}

CSV 和 JSON 文件将具有模式，但开销较小。因此，它甚至可能比 JSON 更好。

在下一节中，我们将看到如何将基于模式的格式作为整体与 Spark 一起使用。

使用 Avro 与 Spark

到目前为止，我们已经看过基于文本的文件。我们使用纯文本、JSON 和 CSV。JSON 和 CSV 比纯文本更好，因为它们携带了一些模式信息。

在本节中，我们将研究一个名为 Avro 的高级模式。将涵盖以下主题：

• 以 Avro 格式保存数据

• 加载 Avro 数据

• 测试

Avro 具有嵌入其中的模式和数据。这是一种二进制格式，不是人类可读的。我们将学习如何以 Avro 格式保存数据，加载数据，然后进行测试。

首先，我们将创建我们的用户交易：

 test("should save and load avro") {
 //given
 import spark.sqlContext.implicits._
 val rdd = spark.sparkContext
     .makeRDD(List(UserTransaction("a", 100), UserTransaction("b", 200)))
     .toDF()

然后我们将进行coalesce并写入 Avro：

 //when
 rdd.coalesce(2)
     .write
     .avro(FileName)

在使用 CSV 时，我们指定了像 CSV 这样的格式，当我们指定 JSON 时，这也是一个格式。但是在 Avro 中，我们有一个方法。这种方法不是标准的 Spark 方法；它来自第三方库。为了具有 Avro 支持，我们需要访问build.sbt并从com.databricks添加spark-avro支持。

然后我们需要导入适当的方法。我们将导入com.databricks.spark.avro._以给我们扩展 Spark DataFrame 的隐式函数：

import com.databricks.spark.avro._

实际上我们正在使用一个 Avro 方法，我们可以看到implicit class接受一个DataFrameWriter类，并以 Spark 格式写入我们的数据。

在我们之前使用的coalesce代码中，我们可以使用write，指定格式，并执行com.databricks.spark.avro类。avro是一个快捷方式，不需要将com.databricks.spark.avro作为整个字符串写入：

//when
 rdd.coalesce(2)
     .write.format(com.databricks.spark.avro)
     .avro(FileName)

简而言之，无需指定格式；只需应用隐式avro方法。

让我们注释掉代码并删除 Avro 以检查它是如何保存的：

// override def afterEach() {
    // val path = Path(FileName)
    // path.deleteRecursively()
// }

如果我们打开transactions.avro文件夹，我们有两部分——part-r-00000和part-r-00001。

第一部分将包含二进制数据。它由许多二进制记录和一些人类可读的数据组成，这就是我们的模式：

我们有两个字段 - user ID，它是一个字符串类型或空值，和name：amount，它是一个整数。作为原始类型，JVM 不能有空值。需要注意的重要事情是，在生产系统中，我们必须保存非常大的数据集，将有成千上万条记录。模式始终在每个文件的第一行。如果我们检查第二部分，我们将看到完全相同的模式，然后是二进制数据。

通常，如果有复杂的模式，我们只有一行或更多行，但仍然是非常少量的数据。

我们可以看到在生成的数据集中，我们有userID和amount：

+------+------+
|userId|amount|
+------+------+
|     a|   100|
|     b|   200|
+------+------+

在上面的代码块中，我们可以看到模式被描绘在文件中。虽然它是一个二进制文件，但我们可以提取它。

在下一节中，我们将研究列格式 - Parquet。

列格式 - Parquet

在本节中，我们将研究第二种基于模式的格式 Parquet。将涵盖以下主题：

• 以 Parquet 格式保存数据

• 加载 Parquet 数据

• 测试

这是一种列格式，因为数据是以列方式存储的，而不是以行方式，就像我们在 JSON、CSV、纯文本和 Avro 文件中看到的那样。

这是一个非常有趣和重要的大数据处理格式，可以加快处理过程。在本节中，我们将专注于向 Spark 添加 Parquet 支持，将数据保存到文件系统中，重新加载数据，然后进行测试。Parquet 与 Avro 类似，因为它提供了一个parquet方法，但这次是一个稍微不同的实现。

在build.sbt文件中，对于 Avro 格式，我们需要添加外部依赖，但对于 Parquet，我们已经在 Spark 中有了该依赖。因此，Parquet 是 Spark 的首选，因为它包含在标准包中。

让我们来看看SaveParquet.scala文件中用于保存和加载 Parquet 文件的逻辑。

首先，我们合并了两个分区，指定了格式，然后指定我们要保存parquet：

package com.tomekl007.chapter_4

import com.databricks.spark.avro._
import com.tomekl007.UserTransaction
import org.apache.spark.sql.SparkSession
import org.scalatest.{BeforeAndAfterEach, FunSuite}

import scala.reflect.io.Path

class SaveParquet extends FunSuite with BeforeAndAfterEach {
  val spark = SparkSession.builder().master("local[2]").getOrCreate()

  private val FileName = "transactions.parquet"

  override def afterEach() {
    val path = Path(FileName)
    path.deleteRecursively()
  }

  test("should save and load parquet") {
    //given
    import spark.sqlContext.implicits._
    val rdd = spark.sparkContext
      .makeRDD(List(UserTransaction("a", 100), UserTransaction("b", 200)))
      .toDF()

    //when
    rdd.coalesce(2)
      .write
      .parquet(FileName)

read方法也实现了完全相同的方法：

    val fromFile = spark.read.parquet(FileName)

    fromFile.show()
    assert(fromFile.count() == 2)
  }

}

让我们开始这个测试，但在此之前，我们将在SaveParquet.scala文件中注释掉以下代码，以查看文件的结构：

//    override def afterEach() {
//    val path = Path(FileName)
//    path.deleteRecursively()
//  } 

创建了一个新的transactions.parquet文件夹，里面有两个部分 - part-r-00000和part-r-00001。但这次，格式完全是二进制的，并且嵌入了一些元数据。

我们嵌入了元数据，还有amount和userID字段，它们是string类型。r-00000部分完全相同，并且嵌入了模式。因此，Parquet 也是一种基于模式的格式。当我们读取数据时，我们可以看到我们有userID和amount列可用。

摘要

在本章中，我们学习了如何以纯文本格式保存数据。我们注意到，当我们没有正确加载数据时，模式信息会丢失。然后我们学习了如何利用 JSON 作为数据格式，并发现 JSON 保留了模式，但它有很多开销，因为模式是针对每条记录的。然后我们了解了 CSV，并发现 Spark 对其有嵌入支持。然而，这种方法的缺点是模式不是关于特定类型的记录，并且需要隐式推断制表符。在本章的最后，我们介绍了 Avro 和 Parquet，它们具有列格式，也嵌入了 Spark。

在下一章中，我们将使用 Spark 的键/值 API。

第十一章：使用 Spark 键/值 API

在本章中，我们将使用 Spark 键/值 API。我们将首先查看可用的键/值对转换。然后，我们将学习如何使用aggregateByKey方法而不是groupBy()方法。稍后，我们将研究键/值对的操作，并查看可用的键/值数据分区器。在本章结束时，我们将实现一个高级分区器，该分区器将能够按范围对我们的数据进行分区。

在本章中，我们将涵盖以下主题：

• 可用的键/值对操作

• 使用aggregateByKey而不是groupBy()

• 键/值对操作

• 可用的键/值数据分区器

• 实现自定义分区器

可用的键/值对操作

在本节中，我们将涵盖以下主题：

• 可用的键/值对转换

• 使用countByKey()

• 了解其他方法

因此，这是我们众所周知的测试，我们将在其中使用键/值对的转换。

首先，我们将为用户A，B，A，B和C创建一个用户交易数组，以某种金额，如下例所示：

 val keysWithValuesList =
 Array(
 UserTransaction("A", 100),
 UserTransaction("B", 4),
 UserTransaction("A", 100001),
 UserTransaction("B", 10),
 UserTransaction("C", 10)
 )

然后，根据以下示例，我们需要按特定字段对数据进行键入：

val keyed = data.keyBy(_.userId)

我们将通过调用keyBy方法并使用userId参数对其进行键入。

现在，我们的数据分配给了keyed变量，其类型为元组。第一个元素是字符串，即userId，第二个元素是UserTransaction。

让我们看一下可用的转换。首先，我们将看看countByKey。

让我们看一下它的实现，如下例所示：

val data = spark.parallelize(keysWithValuesList)
 val keyed = data.keyBy(_.userId)
//when
 val counted = keyed.countByKey()
// keyed.combineByKey()
// keyed.aggregateByKey()
// keyed.foldByKey()
// keyed.groupByKey()
//then
 counted should contain theSameElementsAs Map("B" -> 2, "A" -> 2, "C" -> 1)

这将返回一个Map，键K和Long是一种通用类型，因为它可以是任何类型的键。在本例中，键将是一个字符串。每个返回映射的操作都不是完全安全的。如果您看到返回映射的方法的签名，这表明这些数据将被发送到驱动程序，并且需要适合内存。如果有太多的数据无法适应一个驱动程序的内存，那么我们将耗尽内存。因此，在使用此方法时，我们需要谨慎。

然后，我们执行一个包含与地图相同元素的断言计数，如下例所示：

counted should contain theSameElementsAs Map("B" -> 2, "A" -> 2, "C" -> 1)

B是2，因为我们有两个值。另外，A与C类似，因为它们只有一个值。countByKey()不占用内存，因为它只存储键和计数器。但是，如果键是一个复杂且大的对象，例如具有多个字段的交易，超过两个，那么该映射可能会非常大。

但让我们从下面的例子开始这个测试：

从前面的屏幕截图中，我们可以看到我们的测试通过了。

我们还有一个combineByKey()方法，它将相同键的相同元素组合在一起，并共享负面的aggregateByKey()，能够聚合不同类型。我们有foldByKey，它正在获取当前状态和值，但返回与键的值相同的类型。

我们还有groupByKey()，我们在上一节中了解过。这将根据特定键对所有内容进行分组，并返回键的值迭代器。这也是一个非常占用内存的操作，因此在使用时需要小心。

在下一节中，我们将使用aggregateByKey而不是groupBy。我们将学习groupBy的工作原理并修复其缺陷。

使用aggregateByKey而不是groupBy()

在本节中，我们将探讨为什么我们使用aggregateByKey而不是groupBy。

我们将涵盖以下主题：

• 为什么我们应该避免使用groupByKey

• aggregateByKey给我们的是什么

• 使用aggregateByKey实现逻辑

首先，我们将创建我们的用户交易数组，如下例所示：

 val keysWithValuesList =
 Array(
 UserTransaction("A", 100),
 UserTransaction("B", 4),
 UserTransaction("A", 100001),
 UserTransaction("B", 10),
 UserTransaction("C", 10)
 )

然后，我们将使用parallelize创建一个 RDD，因为我们希望我们的数据按键排序。这在下面的例子中显示：

 val data = spark.parallelize(keysWithValuesList)
 val keyed = data.keyBy(_.userId)

在前面的代码中，我们调用了keyBy来对userId进行操作，以获得付款人、键和用户交易的数据。

让我们假设我们想要聚合，我们想要对相同的键执行一些特定的逻辑，如下面的例子所示：

 val aggregatedTransactionsForUserId = keyed
 .aggregateByKey(amountForUser)(addAmount, mergeAmounts)

这样做的原因可能是选择最大元素、最小元素或计算平均值。aggregateByKey需要接受三个参数，如下面的例子所示：

aggregateByKey(amountForUser)(addAmount, mergeAmounts)

第一个参数是 T 类型的初始参数，定义amountForUser是一个类型为ArrayBuffer的初始参数。这非常重要，因为 Scala 编译器将推断出该类型，并且在这个例子中，参数 1 和 2 需要具有完全相同的类型 T：ArrayBuffer.empty[long]。

下一个参数是一个方法，它接受我们正在处理的当前元素。在这个例子中，transaction: UserTransaction) =>是一个当前交易，也需要带上我们初始化函数的状态，因此这里将是一个数组缓冲区。

它需要与以下代码块中显示的相同类型，因此这是我们的类型 T：

mutable.ArrayBuffer.empty[Long]

在这一点上，我们能够获取任何交易并将其添加到特定状态中。这是以分布式方式完成的。对于一个键，执行在一个执行器上完成，对于完全相同的键，执行在不同的执行器上完成。这是并行进行的，因此对于相同的键将添加多个交易。

现在，Spark 知道，对于完全相同的键，它有多个 T 类型的状态ArrayBuffer，需要合并。因此，我们需要为相同的键mergeAmounts我们的交易。

mergeArgument是一个方法，它接受两个状态，这两个状态都是 T 类型的中间状态，如下面的代码块所示：

 val mergeAmounts = (p1: mutable.ArrayBuffer[Long], p2: mutable.ArrayBuffer[Long]) => p1 ++= p2

在这个例子中，我们想要将释放缓冲区合并成一个数组缓冲区。因此，我们发出p1 ++= p2。这将两个数组缓冲区合并成一个。

现在，我们已经准备好所有参数，我们能够执行aggregateByKey并查看结果是什么样子的。结果是一个字符串和类型 T 的 RDD，ArrayBuffer[long]，这是我们的状态。我们将不再在 RDD 中保留UserTransaction，这有助于减少内存使用。UserTransaction是一个重量级对象，因为它可以有多个字段，在这个例子中，我们只对金额字段感兴趣。因此，这样我们可以减少内存的使用。

下面的例子展示了我们的结果应该是什么样子的：

 aggregatedTransactionsForUserId.collect().toList should contain theSameElementsAs List(
 ("A", ArrayBuffer(100, 100001)),
 ("B", ArrayBuffer(4,10)),
 ("C", ArrayBuffer(10)))

我们应该有一个键A，和一个ArrayBuffer的100和10001，因为这是我们的输入数据。B应该是4和10，最后，C应该是10。

让我们开始测试，检查我们是否已经正确实现了aggregateByKey，如下面的例子所示：

从前面的输出中，我们可以看到它按预期工作。

在下一节中，我们将研究可用于键/值对的操作。

键/值对上的操作

在本节中，我们将研究键/值对上的操作。

我们将涵盖以下主题：

• 检查键/值对上的操作

• 使用collect()

• 检查键/值 RDD 的输出

在本章的第一部分中，我们介绍了可用于键/值对的转换。我们看到它们与 RDD 相比有些不同。此外，对于操作，结果略有不同，但方法名称并没有变化。

因此，我们将使用collect()，并且我们将检查我们对这些键/值对的操作的输出。

首先，我们将根据userId创建我们的交易数组和 RDD，如下面的例子所示：

 val keysWithValuesList =
 Array(
 UserTransaction("A", 100),
 UserTransaction("B", 4),
 UserTransaction("A", 100001),
 UserTransaction("B", 10),
 UserTransaction("C", 10)
 )

我们首先想到的操作是collect()。collect()会取出每个元素并将其分配给结果，因此我们的结果与keyBy的结果非常不同。

我们的结果是一对键，userId和一个值，即UserTransaction。我们可以从下面的例子中看到，我们可以有一个重复的键：

 res should contain theSameElementsAs List(
 ("A",UserTransaction("A",100)),
 ("B",UserTransaction("B",4)),
 ("A",UserTransaction("A",100001)),
 ("B",UserTransaction("B",10)),
 ("C",UserTransaction("C",10))
 )//note duplicated key

在前面的代码中，我们可以看到同一个订单有多个出现。对于一个简单的字符串键，重复并不是很昂贵。然而，如果我们有一个更复杂的键，那么就会很昂贵。

因此，让我们开始这个测试，如下例所示：

从前面的输出中，我们可以看到我们的测试已经通过。要查看其他动作，我们将查看不同的方法。

如果一个方法返回 RDD，比如collect[U] (f: PartialFunction[(String, UserTransaction), U])，这意味着这不是一个动作。如果某些东西返回 RDD，这意味着它不是一个动作。这适用于键/值对。

collect()不会返回 RDD，而是返回数组，因此它是一个动作。count返回long，因此这也是一个动作。countByKey返回 map。如果我们想要reduce我们的元素，那么这是一个动作，但reduceByKey不是一个动作。这就是reduce和reduceByKey之间的重大区别。

我们可以看到根据 RDD，一切都是正常的，因此动作是相同的，差异只在于转换。

在下一节中，我们将看一下键/值数据上可用的分区器。

键/值数据上可用的分区器

我们知道分区和分区器是 Apache Spark 的关键组件。它们影响我们的数据如何分区，这意味着它们影响数据实际驻留在哪些执行器上。如果我们有一个良好的分区器，那么我们将有良好的数据局部性，这将减少洗牌。我们知道洗牌对处理来说是不可取的，因此减少洗牌是至关重要的，因此选择适当的分区器对我们的系统也是至关重要的。

在本节中，我们将涵盖以下主题：

• 检查HashPartitioner

• 检查RangePartitioner

• 测试

我们将首先检查我们的HashPartitioner和RangePartitioner。然后我们将比较它们并使用两个分区器测试代码。

首先，我们将创建一个UserTransaction数组，如下例所示：

 val keysWithValuesList =
 Array(
 UserTransaction("A", 100),
 UserTransaction("B", 4),
 UserTransaction("A", 100001),
 UserTransaction("B", 10),
 UserTransaction("C", 10)
 )

然后我们将使用keyBy（如下例所示），因为分区器将自动处理我们数据的键：

 val keyed = data.keyBy(_.userId)

然后我们将获取键数据的partitioner，如下例所示：

 val partitioner = keyed.partitioner

代码显示partitioner.isEmpty，因为我们还没有定义任何partitioner，因此在这一点上它是空的，如下例所示：

 assert(partitioner.isEmpty)

我们可以使用partitionBy方法指定一个partitioner，如下例所示：

val hashPartitioner = keyed.partitionBy(new HashPartitioner(100))

该方法期望一个partitioner抽象类的实现。我们将有一些实现，但首先，让我们专注于HashPartitioner。

HashPartitioner需要一个分区数，并且有一个分区数。numPartition返回我们的参数，但getPartition会更加复杂，如下例所示：

    def numPartitions: Int = partitions
    def getPartition(key: Any): int = key match {
        case null => 0
        case_ => Utils.nonNegativeMode(key.hashCode, numPartitions)
    }

它首先检查我们的key是否为null。如果是null，它将落在分区号0。如果我们有带有null键的数据，它们都将落在相同的执行器上，正如我们所知，这不是一个好的情况，因为执行器将有很多内存开销，并且它们可能会因为内存异常而失败。

如果key不是null，那么它会从hashCode和分区数中进行nonNegativeMod。它必须是分区数的模数，这样它才能分配到适当的分区。因此，hashCode方法对我们的键非常重要。

如果我们提供了一个自定义的键而不是像整数或字符串这样的原始类型，它有一个众所周知的hashCode，我们需要提供和实现一个适当的hashCode。但最佳实践是使用 Scala 中的case类，因为它们已经为你实现了hashCode和 equals。

我们现在已经定义了partitioner，但partitioner是可以动态更改的。我们可以将我们的partitioner更改为rangePartitioner。rangePartitioner接受 RDD 中的分区。

rangePartitioner更复杂，因为它试图将我们的数据划分为范围，这不像HashPartitioner在获取分区时那样简单。该方法非常复杂，因为它试图均匀地分布我们的数据，并且对将其分布到范围中的逻辑非常复杂。

让我们开始我们的测试，检查我们是否能够正确地分配partitioner，如下所示的输出：

我们的测试已经通过。这意味着，在最初的时候，partitioner是空的，然后我们必须在partitionBy处对 RDD 进行洗牌，还有一个branchPartitioner。但它只显示了我们创建partitioner接口的实例的数值线。

在下一部分，我们将尝试改进它，或者尝试通过实现自定义分区器来调整和玩弄分区器。

实现自定义分区器

在这一部分，我们将实现一个自定义的分区器，并创建一个接受带有范围的解析列表的分区器。如果我们的键落入特定范围，我们将分配列表的分区号索引。

我们将涵盖以下主题：

• 实现自定义分区器

• 实现一个范围分区器

• 测试我们的分区器

我们将根据我们自己的范围分区逻辑来实现范围分区，并测试我们的分区器。让我们从不查看实现的黑盒测试开始。

代码的第一部分与我们已经使用的类似，但这次我们有keyBy数量的数据，如下例所示：

 val keysWithValuesList =
 Array(
 UserTransaction("A", 100),
 UserTransaction("B", 4),
 UserTransaction("A", 100001),
 UserTransaction("B", 10),
 UserTransaction("C", 10)
 )
 val data = spark.parallelize(keysWithValuesList)
 val keyed = data.keyBy(_.amount)

我们按数量进行分组，我们有以下键：100，4，100001，10和10。

然后，我们将创建一个分区器，并将其命名为CustomRangePartitioner，它将接受一个元组列表，如下例所示：

 val partitioned = keyed.partitionBy(new CustomRangePartitioner(List((0,100), (100, 10000), (10000, 1000000))))

第一个元素是从0到100，这意味着如果键在0到100的范围内，它应该进入分区0。因此，有四个键应该落入该分区。下一个分区号的范围是100和10000，因此该范围内的每条记录都应该落入分区号1，包括两端。最后一个范围是10000到1000000元素之间，因此，如果记录在该范围内，它应该落入该分区。如果我们有一个超出范围的元素，那么分区器将因非法参数异常而失败。

让我们看一下下面的例子，展示了我们自定义范围分区器的实现：

class CustomRangePartitioner(ranges: List[(Int,Int)]) extends Partitioner{
 override def numPartitions: Int = ranges.size
override def getPartition(key: Any): Int = {
 if(!key.isInstanceOf[Int]){
 throw new IllegalArgumentException("partitioner works only for Int type")
 }
 val keyInt = key.asInstanceOf[Int]
 val index = ranges.lastIndexWhere(v => keyInt >= v._1 && keyInt <= v._2)
 println(s"for key: $key return $index")
 index
 }
}

它将范围作为元组的参数列表，如下例所示：

(ranges: List[(Int,Int)])

我们的numPartitions应该等于ranges.size，因此分区的数量等于范围的数量。

接下来，我们有getPartition方法。首先，我们的分区器只对整数有效，如下例所示：

if(!key.isInstanceOf[Int])

我们可以看到这是一个整数，不能用于其他类型。出于同样的原因，我们首先需要检查我们的键是否是整数的实例，如果不是，我们会得到一个IllegalArgumentException，因为该分区器只对 int 类型有效。

我们现在可以通过asInstanceOf来测试我们的keyInt。完成后，我们可以遍历范围，并在索引在谓词之间时取最后一个范围。我们的谓词是一个元组v，应该如下所示：

 val index = ranges.lastIndexWhere(v => keyInt >= v._1 && keyInt <= v._2)

KeyInt应该大于或等于v._1，即元组的第一个元素，但也应该小于第二个元素v._2。

范围的起始是v._1，范围的结束是v._2，因此我们可以检查我们的元素是否在范围内。

最后，我们将打印我们在调试目的中找到的键的索引，并返回索引，这将是我们的分区。如下例所示：

println(s"for key: $key return $index")

让我们开始下面的测试：

我们可以看到对于键100001，代码返回了预期的分区号2。对于键100返回分区一，对于10，4，10返回分区零，这意味着我们的代码按预期工作。

摘要

在本章中，我们首先看到了关于键/值对的转换操作。然后我们学习了如何使用aggregateByKey而不是groupBy。我们还涵盖了关于键/值对的操作。之后，我们看了一下可用的分区器，比如rangePartitioner和HashPartition在键/值数据上。在本章结束时，我们已经实现了我们自定义的分区器，它能够根据范围的起始和结束来分配分区，以便学习目的。

在下一章中，我们将学习如何测试我们的 Spark 作业和 Apache Spark 作业。

第十二章：测试 Apache Spark 作业

在本章中，我们将测试 Apache Spark 作业，并学习如何将逻辑与 Spark 引擎分离。

我们将首先对我们的代码进行单元测试，然后在 SparkSession 中进行集成测试。之后，我们将使用部分函数模拟数据源，然后学习如何利用 ScalaCheck 进行基于属性的测试以及 Scala 中的类型。在本章结束时，我们将在不同版本的 Spark 中执行测试。

在本章中，我们将涵盖以下主题：

• 将逻辑与 Spark 引擎分离-单元测试

• 使用 SparkSession 进行集成测试

• 使用部分函数模拟数据源

• 使用 ScalaCheck 进行基于属性的测试

• 在不同版本的 Spark 中进行测试

将逻辑与 Spark 引擎分离-单元测试

让我们从将逻辑与 Spark 引擎分离开始。

在本节中，我们将涵盖以下主题：

• 创建具有逻辑的组件

• 该组件的单元测试

• 使用模型类的案例类进行领域逻辑

让我们先看逻辑，然后是简单的测试。

因此，我们有一个BonusVerifier对象，只有一个方法quaifyForBonus，它接受我们的userTransaction模型类。根据以下代码中的登录，我们加载用户交易并过滤所有符合奖金资格的用户。首先，我们需要测试它以创建一个 RDD 并对其进行过滤。我们需要创建一个 SparkSession，并为模拟 RDD 或 DataFrame 创建数据，然后测试整个 Spark API。由于这涉及逻辑，我们将对其进行隔离测试。逻辑如下：

package com.tomekl007.chapter_6
import com.tomekl007.UserTransaction
object BonusVerifier {
 private val superUsers = List("A", "X", "100-million")
def qualifyForBonus(userTransaction: UserTransaction): Boolean = {
 superUsers.contains(userTransaction.userId) && userTransaction.amount > 100
 }
}

我们有一个超级用户列表，其中包括A、X和100-million用户 ID。如果我们的userTransaction.userId在superUsers列表中，并且userTransaction.amount高于100，那么用户就有资格获得奖金；否则，他们就没有资格。在现实世界中，奖金资格逻辑将更加复杂，因此非常重要的是对逻辑进行隔离测试。

以下代码显示了我们使用userTransaction模型的测试。我们知道我们的用户交易包括userId和amount。以下示例显示了我们的领域模型对象，它在 Spark 执行集成测试和我们的单元测试之间共享，与 Spark 分开：

package com.tomekl007

import java.util.UUID

case class UserData(userId: String , data: String)

case class UserTransaction(userId: String, amount: Int)

case class InputRecord(uuid: String = UUID.*randomUUID()*.toString(), userId: String)

我们需要为用户 ID X 和金额101创建我们的UserTransaction，如下例所示：

package com.tomekl007.chapter_6
import com.tomekl007.UserTransaction
import org.scalatest.FunSuite
class SeparatingLogic extends FunSuite {
test("test complex logic separately from spark engine") {
 //given
 val userTransaction = UserTransaction("X", 101)
//when
 val res = BonusVerifier.qualifyForBonus(userTransaction)
//then
 assert(res)
 }
}

然后我们将userTransaction传递给qualifyForBonus，结果应该是true。这个用户应该有资格获得奖金，如下输出所示：

现在，让我们为负面用例编写一个测试，如下所示：

test(testName = "test complex logic separately from spark engine - non qualify") {
 //given
 val userTransaction = UserTransaction("X", 99)
//when
 val res = BonusVerifier.*qualifyForBonus*(userTransaction)
//then
 assert(!res)
 }

在这里，我们有一个用户X，花费99，所以我们的结果应该是 false。当我们验证我们的代码时，我们可以看到从以下输出中，我们的测试已经通过了：

我们已经涵盖了两种情况，但在现实世界的场景中，还有更多。例如，如果我们想测试指定userId不在这个超级用户列表中的情况，我们有一个花了很多钱的some_new_user，在我们的案例中是100000，我们得到以下结果：

test(testName = "test complex logic separately from spark engine - non qualify2") {
 //given
 val userTransaction = UserTransaction("some_new_user", 100000)
//when
 val res = BonusVerifier.*qualifyForBonus*(userTransaction)
//then
 assert(!res)
 }

假设它不应该符合条件，因此这样的逻辑有点复杂。因此，我们以单元测试的方式进行测试：

我们的测试非常快，因此我们能够检查一切是否按预期工作，而无需引入 Spark。在下一节中，我们将使用 SparkSession 进行集成测试来更改逻辑。

使用 SparkSession 进行集成测试

现在让我们学习如何使用 SparkSession 进行集成测试。

在本节中，我们将涵盖以下主题：

• 利用 SparkSession 进行集成测试

• 使用经过单元测试的组件

在这里，我们正在创建 Spark 引擎。以下行对于集成测试至关重要：

 val spark: SparkContext = SparkSession.builder().master("local[2]").getOrCreate().sparkContext

创建一个轻量级对象并不是一件简单的事情。SparkSession 是一个非常重的对象，从头开始构建它是一项昂贵的操作，从资源和时间的角度来看。与上一节的单元测试相比，诸如创建 SparkSession 的测试将花费更多的时间。

出于同样的原因，我们应该经常使用单元测试来转换所有边缘情况，并且仅在逻辑的较小部分，如资本边缘情况时才使用集成测试。

以下示例显示了我们正在创建的数组：

 val keysWithValuesList =
 Array(
 UserTransaction("A", 100),
 UserTransaction("B", 4),
 UserTransaction("A", 100001),
 UserTransaction("B", 10),
 UserTransaction("C", 10)
 )

以下示例显示了我们正在创建的 RDD：

 val data = spark.parallelize(keysWithValuesList)

这是 Spark 第一次参与我们的集成测试。创建 RDD 也是一个耗时的操作。与仅创建数组相比，创建 RDD 真的很慢，因为它也是一个重量级对象。

我们现在将使用我们的data.filter来传递一个qualifyForBonus函数，如下例所示：

 val aggregatedTransactionsForUserId = data.filter(BonusVerifier.qualifyForBonus)

这个函数已经经过单元测试，所以我们不需要考虑所有边缘情况，不同的 ID，不同的金额等等。我们只是创建了一些 ID 和一些金额来测试我们整个逻辑链是否按预期工作。

应用了这个逻辑之后，我们的输出应该类似于以下内容：

 UserTransaction("A", 100001)

让我们开始这个测试，检查执行单个集成测试需要多长时间，如下输出所示：

执行这个简单测试大约需要646 毫秒。

如果我们想要覆盖每一个边缘情况，与上一节的单元测试相比，值将乘以数百倍。让我们从三个边缘情况开始这个单元测试，如下输出所示：

我们可以看到我们的测试只花了18 毫秒，这意味着即使我们覆盖了三个边缘情况，与只有一个情况的集成测试相比，速度快了 20 倍。

在这里，我们覆盖了许多逻辑，包括数百个边缘情况，我们可以得出结论，尽可能低的级别进行单元测试是非常明智的。

在下一节中，我们将使用部分函数来模拟数据源。

使用部分函数模拟数据源

在本节中，我们将涵盖以下主题：

• 创建一个从 Hive 读取数据的 Spark 组件

• 模拟组件

• 测试模拟组件

假设以下代码是我们的生产线：

 ignore("loading data on prod from hive") {
 UserDataLogic.loadAndGetAmount(spark, HiveDataLoader.loadUserTransactions)
 }

在这里，我们使用UserDataLogic.loadAndGetAmount函数，它需要加载我们的用户数据交易并获取交易的金额。这个方法需要两个参数。第一个参数是sparkSession，第二个参数是sparkSession的provider，它接受SparkSession并返回DataFrame，如下例所示：

object UserDataLogic {
  def loadAndGetAmount(sparkSession: SparkSession, provider: SparkSession => DataFrame): DataFrame = {
    val df = provider(sparkSession)
    df.select(df("amount"))
  }
}

对于生产，我们将加载用户交易，并查看HiveDataLoader组件只有一个方法，sparkSession.sql和("select * from transactions")，如下代码块所示：

object HiveDataLoader {
 def loadUserTransactions(sparkSession: SparkSession): DataFrame = {
 sparkSession.sql("select * from transactions")
 }
}

这意味着该函数去 Hive 检索我们的数据并返回一个 DataFrame。根据我们的逻辑，它执行了返回 DataFrame 的provider，然后从 DataFrame 中选择amount。

这个逻辑并不简单，因为我们的 SparkSession provider在生产中与外部系统进行交互。因此，我们可以创建一个如下的函数：

UserDataLogic.loadAndGetAmount(spark, HiveDataLoader.loadUserTransactions)

让我们看看如何测试这样一个组件。首先，我们将创建一个用户交易的 DataFrame，这是我们的模拟数据，如下例所示：

 val df = spark.sparkContext
 .makeRDD(List(UserTransaction("a", 100), UserTransaction("b", 200)))
 .toDF()

然而，我们需要将数据保存到 Hive 中，嵌入它，然后启动 Hive。

由于我们使用了部分函数，我们可以将部分函数作为第二个参数传递，如下例所示：

val res = UserDataLogic.loadAndGetAmount(spark, _ => df)

第一个参数是spark，但这次我们的方法中没有使用它。第二个参数是一个接受 SparkSession 并返回 DataFrame 的方法。

然而，我们的执行引擎、架构和代码并不考虑这个 SparkSession 是否被使用，或者是否进行了外部调用；它只想返回 DataFrame。我们可以使用_作为我们的第一个参数，因为它被忽略，只返回 DataFrame 作为返回类型。

因此我们的loadAndGetAmount将获得一个模拟 DataFrame，这是我们创建的 DataFrame。

但是，对于所示的逻辑，它是透明的，不考虑 DataFrame 是来自 Hive、SQL、Cassandra 还是其他任何来源，如下例所示：

 val df = provider(sparkSession)
 df.select(df("amount"))

在我们的例子中，df来自我们为测试目的创建的内存。我们的逻辑继续并选择了数量。

然后，我们展示我们的列，res.show()，并且该逻辑应该以一个列的数量结束。让我们开始这个测试，如下例所示：

我们可以从上面的例子中看到，我们的结果 DataFrame 在100和200值中有一个列的数量。这意味着它按预期工作，而无需启动嵌入式 Hive。关键在于使用提供程序而不是在逻辑中嵌入我们的选择开始。

在下一节中，我们将使用 ScalaCheck 进行基于属性的测试。

使用 ScalaCheck 进行基于属性的测试

在本节中，我们将涵盖以下主题：

• 基于属性的测试

• 创建基于属性的测试

让我们看一个简单的基于属性的测试。在定义属性之前，我们需要导入一个依赖项。我们还需要一个 ScalaCheck 库的依赖项，这是一个用于基于属性的测试的库。

在上一节中，每个测试都扩展了FunSuite。我们使用了功能测试，但是必须显式提供参数。在这个例子中，我们扩展了来自 ScalaCheck 库的Properties，并测试了StringType，如下所示：

object PropertyBasedTesting extends Properties("StringType")

我们的 ScalaCheck 将为我们生成一个随机字符串。如果我们为自定义类型创建基于属性的测试，那么 ScalaCheck 是不知道的。我们需要提供一个生成器，它将生成该特定类型的实例。

首先，让我们以以下方式定义我们字符串类型的第一个属性：

property("length of strings") = forAll { (a: String, b: String) =>
 a.length + b.length >= a.length
 }

forAll是 ScalaCheck 属性的一个方法。我们将在这里传递任意数量的参数，但它们需要是我们正在测试的类型。

假设我们想要获得两个随机字符串，并且在这些字符串中，不变性应该被感知。

如果我们将字符串a的长度加上字符串b的长度，那么它们的总和应该大于或等于a.length，因为如果b是0，那么它们将相等，如下例所示：

a.length + b.length >= a.length

然而，这是string的不变性，对于每个输入字符串，它应该是true。

我们正在定义的第二个属性更复杂，如下代码所示：

property("creating list of strings") = forAll { (a: String, b: String, c: String) =>
 List(a,b,c).map(_.length).sum == a.length + b.length + c.length
 }

在上面的代码中，我们要求 ScalaCheck 运行时引擎这次共享三个字符串，即a、b和c。当我们创建一个字符串列表时，我们将测试这个。

在这里，我们正在创建一个字符串列表，即a、b、c，如下代码所示：

List(a,b,c)

当我们将每个元素映射到length时，这些元素的总和应该等于通过长度添加所有元素。在这里，我们有a.length + b.length + c.length，我们将测试集合 API，以检查映射和其他函数是否按预期工作。

让我们开始这个基于属性的测试，以检查我们的属性是否正确，如下例所示：

我们可以看到string的StringType.length属性通过并执行了100次测试。100次测试被执行可能会让人惊讶，但让我们尝试看看通过以下代码传递了什么参数：

println(s"a: $a, b: $b")

我们将打印a参数和b参数，并通过测试以下输出来重试我们的属性：

我们可以看到生成了许多奇怪的字符串，因此这是一个我们无法事先创建的边缘情况。基于属性的测试将创建一个非常奇怪的唯一代码，这不是一个合适的字符串。因此，这是一个用于测试我们的逻辑是否按预期针对特定类型工作的好工具。

在下一节中，我们将在不同版本的 Spark 中进行测试。

在不同版本的 Spark 中进行测试

在本节中，我们将涵盖以下主题：

• 将组件更改为与 Spark pre-2.x 一起使用

• Mock 测试 pre-2.x

• RDD 模拟测试

让我们从本章第三节开始，模拟数据源——使用部分函数模拟数据源。

由于我们正在测试UserDataLogic.loadAndGetAmount，请注意一切都在 DataFrame 上操作，因此我们有一个 SparkSession 和 DataFrame。

现在，让我们将其与 Spark pre-2.x 进行比较。我们可以看到这一次我们无法使用 DataFrame。假设以下示例显示了我们在以前的 Spark 中的逻辑：

test("mock loading data from hive"){
 //given
 import spark.sqlContext.implicits._
 val df = spark.sparkContext
 .makeRDD(List(UserTransaction("a", 100), UserTransaction("b", 200)))
 .toDF()
 .rdd
//when
 val res = UserDataLogicPre2.loadAndGetAmount(spark, _ => df)
//then
 println(res.collect().toList)
 }
}

我们可以看到这一次我们无法使用 DataFrame。

在前面的部分中，loadAndGetAmount正在接受spark和 DataFrame，但在下面的示例中，DataFrame 是一个 RDD，不再是 DataFrame，因此我们传递了一个rdd：

 val res = UserDataLogicPre2.loadAndGetAmount(spark, _ => rdd)

然而，我们需要为 Spark 创建一个不同的UserDataLogicPre2，它接受 SparkSession 并在映射整数的 RDD 之后返回一个 RDD，如下例所示：

object UserDataLogicPre2 {
 def loadAndGetAmount(sparkSession: SparkSession, provider: SparkSession => RDD[Row]): RDD[Int] = {
 provider(sparkSession).map(_.getAsInt)
 }
}
object HiveDataLoaderPre2 {
 def loadUserTransactions(sparkSession: SparkSession): RDD[Row] = {
 sparkSession.sql("select * from transactions").rdd
 }
}

在前面的代码中，我们可以看到provider正在执行我们的提供程序逻辑，映射每个元素，将其作为int获取。然后，我们得到了金额。Row是一个可以有可变数量参数的泛型类型。

在 Spark pre-2.x 中，我们没有SparkSession，因此需要使用SparkContext并相应地更改我们的登录。

总结

在本章中，我们首先学习了如何将逻辑与 Spark 引擎分离。然后，我们查看了一个在没有 Spark 引擎的情况下经过良好测试的组件，并使用 SparkSession 进行了集成测试。为此，我们通过重用已经经过良好测试的组件创建了一个 SparkSession 测试。通过这样做，我们不必在集成测试中涵盖所有边缘情况，而且我们的测试速度更快。然后，我们学习了如何利用部分函数在测试阶段提供模拟数据。我们还介绍了 ScalaCheck 用于基于属性的测试。在本章结束时，我们已经在不同版本的 Spark 中测试了我们的代码，并学会了将 DataFrame 模拟测试更改为 RDD。

在下一章中，我们将学习如何利用 Spark GraphX API。

第十三章：利用 Spark GraphX API

在本章中，我们将学习如何从数据源创建图。然后，我们将使用 Edge API 和 Vertex API 进行实验。在本章结束时，您将知道如何计算顶点的度和 PageRank。

在本章中，我们将涵盖以下主题：

• 从数据源创建图

• 使用 Vertex API

• 使用 Edge API

• 计算顶点的度

• 计算 PageRank

从数据源创建图

我们将创建一个加载器组件，用于加载数据，重新审视图格式，并从文件加载 Spark 图。

创建加载器组件

graph.g文件包含顶点到顶点的结构。在下面的graph.g文件中，如果我们将1对齐到2，这意味着顶点 ID1和顶点 ID2之间有一条边。第二行表示从顶点 ID1到顶点 ID3有一条边，然后从2到3，最后从3到5：

1  2
1  3
2  3
3  5

我们将取graph.g文件，加载它，并查看它将如何在 Spark 中提供结果。首先，我们需要获取我们的graph.g文件的资源。我们将使用getClass.getResource()方法来获取它的路径，如下所示：

package com.tomekl007.chapter_7

import org.apache.spark.SparkContext
import org.apache.spark.sql.SparkSession
import org.scalatest.FunSuite

class CreatingGraph extends FunSuite {
  val spark: SparkContext = SparkSession.builder().master("local[2]").getOrCreate().sparkContext

  test("should load graph from a file") {
    //given
    val path = getClass.getResource("/graph.g").getPath

重新审视图格式

接下来，我们有GraphBuilder方法，这是我们自己的组件：

    //when
    val graph = GraphBuilder.loadFromFile(spark, path)

以下是我们的GraphBuilder.scala文件，用于我们的GraphBuilder方法：

package com.tomekl007.chapter_7

import org.apache.spark.SparkContext
import org.apache.spark.graphx.{Graph, GraphLoader}

object GraphBuilder {

  def loadFromFile(sc: SparkContext, path: String): Graph[Int, Int] = {
    GraphLoader.edgeListFile(sc, path)
  }
}

它使用了org.apache.spark.graphx.{Graph, GraphLoader}包中的GraphLoader类，并且我们指定了格式。

这里指定的格式是edgeListFile。我们传递了sc参数，即SparkContext和path参数，其中包含文件的路径。得到的图将是Graph [Int, Int]，我们将使用它作为我们顶点的标识符。

从文件加载 Spark

一旦我们得到了结果图，我们可以将spark和path参数传递给我们的GraphBuilder.loadFromFile()方法，此时，我们将得到一个Graph [Int, Int]的构造图，如下所示：

  val graph = GraphBuilder.loadFromFile(spark, path)

迭代和验证我们的图是否被正确加载，我们将使用图中的三元组，它们是一对顶点到顶点，也是这些顶点之间的边。我们将看到图的结构是否被正确加载：

    //then
    graph.triplets.foreach(println(_))

最后，我们断言我们得到了4个三元组（如前面在创建加载器组件部分中所示，我们从graph.g文件中有四个定义）：

    assert(graph.triplets.count() == 4)
  }

}

我们将开始测试并查看我们是否能够正确加载我们的图。

我们得到了以下输出。这里，我们有(2, 1)，(3, 1)，(3,1)，(5,1)，(1,1)，(2,1)，(1,1)和(3,1)：

因此，根据输出的图，我们能够使用 Spark 重新加载我们的图。

使用 Vertex API

在这一部分，我们将使用边来构建图。我们将学习使用 Vertex API，并利用边的转换。

使用顶点构建图

构建图不是一项简单的任务；我们需要提供顶点和它们之间的边。让我们专注于第一部分。第一部分包括我们的users，users是一个VertexId和String的 RDD，如下所示：

package com.tomekl007.chapter_7

import org.apache.spark.SparkContext
import org.apache.spark.graphx.{Edge, Graph, VertexId}
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.SparkSession
import org.scalatest.FunSuite

class VertexAPI extends FunSuite {
  val spark: SparkContext = SparkSession.builder().master("local[2]").getOrCreate().sparkContext

  test("Should use Vertex API") {
    //given
    val users: RDD[(VertexId, (String))] =
      spark.parallelize(Array(
        (1L, "a"),
        (2L, "b"),
        (3L, "c"),
        (4L, "d")
      ))

VertexId是long类型；这只是Long的type别名：

type VertexID = Long

但由于我们的图有时包含大量内容，VertexId应该是唯一的且非常长的数字。我们的顶点 RDD 中的每个顶点都应该有一个唯一的VertexId。与顶点关联的自定义数据可以是任何类，但我们将选择使用String类来简化。首先，我们创建一个 ID 为1的顶点和字符串数据a，下一个 ID 为2的顶点和字符串数据b，下一个 ID 为3的顶点和字符串数据c，以及 ID 为4的数据和字符串d，如下所示：

    val users: RDD[(VertexId, (String))] =
      spark.parallelize(Array(
        (1L, "a"),
        (2L, "b"),
        (3L, "c"),
        (4L, "d")
      ))

仅从顶点创建图是正确的，但并不是非常有用。图是查找数据之间关系的最佳方式，这就是为什么图是社交网络的主要构建块。

创建夫妻关系

在这一部分，我们将创建顶点之间的夫妻关系和边缘。在这里，我们将有一个关系，即Edge。Edge是来自org.apache.spark.graphx包的一个样例类。它稍微复杂一些，因为我们需要指定源顶点 ID 和目标顶点 ID。我们想要指定顶点 ID1和2有一个关系，所以让我们为这个关系创建一个标签。在下面的代码中，我们将指定顶点 ID1和 ID2为friend，然后我们还将指定顶点 ID1和 ID3也为friend。最后，顶点 ID2和 ID4将是wife：

    val relationships =
      spark.parallelize(Array(
        Edge(1L, 2L, "friend"),
        Edge(1L, 3L, "friend"),
        Edge(2L, 4L, "wife")
      ))

此外，标签可以是任何类型-它不需要是String类型；我们可以输入我们想要的内容并传递它。一旦我们有了我们的顶点、用户和边缘关系，我们就可以创建一个图。我们使用Graph类的apply方法来构建我们的 Spark GraphX 图。我们需要传递users、VertexId和relationships，如下所示：

返回的graph是一个 RDD，但它是一个特殊的 RDD：

    val graph = Graph(users, relationships)

当我们转到Graph类时，我们会看到Graph类有一个顶点的 RDD 和一个边缘的 RDD，所以Graph类是两个 RDD 的伴生对象，如下截图所示：

我们可以通过发出一些方法来获取vertices和edges的基础 RDD。例如，如果要获取所有顶点，我们可以映射所有顶点，我们将获取属性和VertexId。在这里，我们只对属性感兴趣，我们将其转换为大写，如下所示：

    val res = graph.mapVertices((_, att) => att.toUpperCase())

以下是属性：

    val users: RDD[(VertexId, (String))] =
      spark.parallelize(Array(
        (1L, "a"),
        (2L, "b"),
        (3L, "c"),
        (4L, "d")
      ))

一旦我们将其转换为大写，我们可以收集所有顶点并执行toList()，如下所示：

    println(res.vertices.collect().toList)
  }

}

我们可以看到，在对值应用转换后，我们的图具有以下顶点：

使用 Edge API

在这一部分，我们将使用 Edge API 构建图。我们还将使用顶点，但这次我们将专注于边缘转换。

使用边缘构建图

正如我们在前面的部分中看到的，我们有边缘和顶点，这是一个 RDD。由于这是一个 RDD，我们可以获取一个边缘。我们有许多可用于普通 RDD 的方法。我们可以使用max方法、min方法、sum方法和所有其他操作。我们将应用reduce方法，因此reduce方法将获取两个边缘，我们将获取e1、e2，并对其执行一些逻辑。

e1边缘是一个具有属性、目的地和源的边缘，如下截图所示：

由于边缘将两个顶点链接在一起，我们可以在这里执行一些逻辑。例如，如果e1边缘属性等于friend，我们希望使用filter操作提升一个边缘。因此，filter方法只获取一个边缘，然后如果边缘e1是friend，它将被自动感知。我们可以看到最后我们可以collect它并执行toList，以便 Spark 上的 API 可供我们使用。以下代码将帮助我们实现我们的逻辑：

import org.apache.spark.SparkContext
import org.apache.spark.graphx.{Edge, Graph, VertexId}
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.SparkSession
import org.scalatest.FunSuite

class EdgeAPI extends FunSuite {
  val spark: SparkContext = SparkSession.builder().master("local[2]").getOrCreate().sparkContext

  test("Should use Edge API") {
    //given
    val users: RDD[(VertexId, (String))] =
      spark.parallelize(Array(
        (1L, "a"),
        (2L, "b"),
        (3L, "c"),
        (4L, "d")
      ))

    val relationships =
      spark.parallelize(Array(
        Edge(1L, 2L, "friend"),
        Edge(1L, 3L, "friend"),
        Edge(2L, 4L, "wife")
      ))

    val graph = Graph(users, relationships)

    //when
 val resFromFilter = graph.edges.filter((e1) => e1.attr == "friend").collect().toList
 println(resFromFilter)

它还具有标准 RDD 的一些方法。例如，我们可以执行一个 map edge，它将获取一个边缘，我们可以获取一个属性，并将每个标签映射为大写，如下所示：

    val res = graph.mapEdges(e => e.attr.toUpperCase)

在图上，我们还可以执行边缘分组。边缘分组类似于GROUP BY，但仅适用于边缘。

输入以下命令以打印线路映射边缘：

    println(res.edges.collect().toList)

让我们开始我们的代码。我们可以在输出中看到，我们的代码已经过滤了wife边缘-我们只能感知从顶点 ID1到 ID2的friend边缘，以及从顶点 ID1到 ID3的边缘，并将边缘映射如下截图所示：

计算顶点的度

在本节中，我们将涵盖总度数，然后将其分为两部分——入度和出度——并且我们将了解这在代码中是如何工作的。

对于我们的第一个测试，让我们构建我们已经了解的图：

package com.tomekl007.chapter_7

import org.apache.spark.SparkContext
import org.apache.spark.graphx.{Edge, Graph, VertexId}
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.SparkSession
import org.scalatest.FunSuite
import org.scalatest.Matchers._

class CalculateDegreeTest extends FunSuite {
  val spark: SparkContext = SparkSession.builder().master("local[2]").getOrCreate().sparkContext

  test("should calculate degree of vertices") {
    //given
    val users: RDD[(VertexId, (String))] =
      spark.parallelize(Array(
        (1L, "a"),
        (2L, "b"),
        (3L, "c"),
        (4L, "d")
      ))

    val relationships =
      spark.parallelize(Array(
        Edge(1L, 2L, "friend"),
        Edge(1L, 3L, "friend"),
        Edge(2L, 4L, "wife")
      ))

我们可以使用degrees方法获得度。degrees方法返回VertexRDD，因为degrees是一个顶点：

    val graph = Graph(users, relationships)

    //when
    val degrees = graph.degrees.collect().toList

结果如下：

    //then
    degrees should contain theSameElementsAs List(
      (4L, 1L),
      (2L, 2L),
      (1L, 2L),
      (3L, 1L)
    )
  }

上面的代码解释了对于VertexId 4L实例，只有一个关系，因为2L和4L之间存在关系。

然后，对于VertexId 2L实例，有两个，分别是1L, 2L和2L, 4L。对于VertexId 1L实例，有两个，分别是1L, 2L和1L, 3L，对于VertexId 3L，只有一个关系，即1L和3L之间。通过这种方式，我们可以检查我们的图是如何耦合的，以及有多少关系。我们可以通过对它们进行排序来找出哪个顶点最知名，因此我们可以看到我们的测试在下面的截图中通过了。

入度

入度告诉我们有多少个顶点进入第二个顶点，但反之则不然。这次，我们可以看到对于VertexId 2L实例，只有一个入站顶点。我们可以看到2L与1L有关系，3L也与1L有关系，4L与1L有关系。在下面的结果数据集中，将没有VertexId 1L的数据，因为1L是输入。所以，1L只会是一个源，而不是目的地：

  test("should calculate in-degree of vertices") {
    //given
    val users: RDD[(VertexId, (String))] =
      spark.parallelize(Array(
        (1L, "a"),
        (2L, "b"),
        (3L, "c"),
        (4L, "d")
      ))

    val relationships =
      spark.parallelize(Array(
        Edge(1L, 2L, "friend"),
        Edge(1L, 3L, "friend"),
        Edge(2L, 4L, "wife")
      ))

    val graph = Graph(users, relationships)

    //when
    val degrees = graph.inDegrees.collect().toList

    //then
    degrees should contain theSameElementsAs List(
      (2L, 1L),
      (3L, 1L),
      (4L, 1L)
    )
  }

入度的前面特征是一个非常有用的属性。当我们无法找出哪些页面非常重要，因为它们通过页面而不是从页面链接时，我们使用入度。

通过运行这个测试，我们可以看到它按预期工作：

出度

出度解释了有多少个顶点出去。这次，我们将计算边缘、关系的源，而不是目的地，就像我们在入度方法中所做的那样。

为了获得出度，我们将使用以下代码：

val degrees = graph.outDegrees.collect().toList

outDegrees方法包含RDD和VertexRDD，我们使用collect和toList方法将其收集到列表中。

在这里，VertexId 1L应该有两个出站顶点，因为1L, 2L和1L, 3L之间存在关系：

  test("should calculate out-degree of vertices") {
    //given
    val users: RDD[(VertexId, (String))] =
      spark.parallelize(Array(
        (1L, "a"),
        (2L, "b"),
        (3L, "c"),
        (4L, "d")
      ))

    val relationships =
      spark.parallelize(Array(
        Edge(1L, 2L, "friend"),
        Edge(1L, 3L, "friend"),
        Edge(2L, 4L, "wife")
      ))

    val graph = Graph(users, relationships)

    //when
    val degrees = graph.outDegrees.collect().toList

    //then
    degrees should contain theSameElementsAs List(
      (1L, 2L),
      (2L, 1L)
    )
  }

}

另外，VertexId 2L应该有一个出站顶点，因为2L和4L之间存在关系，而反之则不然，如前面的代码所示。

我们将运行这个测试并得到以下输出：

计算 PageRank

在本节中，我们将加载关于用户的数据并重新加载关于他们关注者的数据。我们将使用图形 API 和我们的数据结构，并计算 PageRank 来计算用户的排名。

首先，我们需要加载edgeListFile，如下所示：

package com.tomekl007.chapter_7

import org.apache.spark.graphx.GraphLoader
import org.apache.spark.sql.SparkSession
import org.scalatest.FunSuite
import org.scalatest.Matchers._

class PageRankTest extends FunSuite {
  private val sc = SparkSession.builder().master("local[2]").getOrCreate().sparkContext

  test("should calculate page rank using GraphX API") {
    //given
    val graph = GraphLoader.edgeListFile(sc, getClass.getResource("/pagerank/followers.txt").getPath)

我们有一个followers.txt文件；以下截图显示了文件的格式，与我们在创建加载器组件部分看到的文件类似：

我们可以看到每个顶点 ID 之间存在关系。因此，我们从followers.txt文件加载graph，然后发出 PageRank。我们将需要vertices，如下所示：

    val ranks = graph.pageRank(0.0001).vertices

PageRank 将计算我们的顶点之间的影响和关系。

加载和重新加载关于用户和关注者的数据

为了找出哪个用户有哪个名字，我们需要加载users.txt文件。users.txt文件将VertexId分配给用户名和自己的名字。我们使用以下代码：

    val users = sc.textFile(getClass.getResource("/pagerank/users.txt").getPath).map { line =>

以下是users.txt文件：

我们在逗号上拆分，第一组是我们的整数，它将是顶点 ID，然后fields(1)是顶点的名称，如下所示：

      val fields = line.split(",")
      (fields(0).toLong, fields(1))
    }

接下来，我们将使用join将users与ranks连接起来。我们将使用用户的VertexId通过用户的username和rank来join users。一旦我们有了这些，我们就可以按rank对所有内容进行排序，所以我们将取元组的第二个元素，并且应该按sortBy ((t) =>t.2进行排序。在文件的开头，我们将拥有影响力最大的用户：

    //when
 val rankByUsername = users.join(ranks).map {
      case (_, (username, rank)) => (username, rank)
    }.sortBy((t) => t._2, ascending = false)
      .collect()
      .toList

我们将打印以下内容并按rankByUsername进行排序：

    println(rankByUsername)
    //then
    rankByUsername.map(_._1) should contain theSameElementsInOrderAs List(
      "BarackObama",
      "ladygaga",
      "odersky",
      "jeresig",
      "matei_zaharia",
      "justinbieber"
    )
  }

}

如果我们跳过sortBy方法，Spark 不保证元素的任何排序；为了保持排序，我们需要使用sortBy方法。

在运行代码后，我们得到以下输出：

当我们开始运行这个测试时，我们可以看到 GraphX PageRank 是否能够计算出我们用户的影响力。我们得到了前面截图中显示的输出，其中BarackObama的影响力最大为1.45，然后是ladygaga，影响力为1.39，odersky为1.29，jeresig为0.99，matai_zaharia为0.70，最后是justinbieber，影响力为0.15。

根据前面的信息，我们能够用最少的代码计算复杂的算法。

总结

在本章中，我们深入研究了转换和操作，然后学习了 Spark 的不可变设计。我们研究了如何避免洗牌以及如何减少运营成本。然后，我们看了如何以正确的格式保存数据。我们还学习了如何使用 Spark 键/值 API 以及如何测试 Apache Spark 作业。之后，我们学习了如何从数据源创建图形，然后研究并尝试了边缘和顶点 API。我们学习了如何计算顶点的度。最后，我们看了 PageRank 以及如何使用 Spark GraphicX API 进行计算。

标签：数据分析,指南,val,PySpark,RDD,使用,Spark,数据,我们
From： https://www.cnblogs.com/apachecn/p/18203718