【ML】第 2 章：PySpark 简介

许多书都是关于 Apache Spark 的。它们都深入介绍了它是什么、何时使用以及如何使用。本章将带您快速了解 PySpark——足以让您对本书的其余部分感到满意。要深入了解 Spark 本身，请获取一份Spark：权威指南。

让我们从头开始。Spark到底是什么？

Apache Spark 最初于 2009 年在加州大学伯克利分校开发，是一个用于大数据和机器学习的分析引擎。自 Spark 发布以来，许多行业的企业都迅速采用了它。包括 Netflix、雅虎和 eBay 在内的几家巨头都大规模部署了 Spark，在数千个节点的集群上处理 EB 级数据。这很快使 Spark 成为最大的大数据开源社区，拥有来自 250 多个组织的 1000 多名贡献者。

那么，什么是 PySpark ？

虽然 Spark 本身是用 Scala 和 Java 编写的，但 PySpark 是一个允许我们使用 Python 与 Spark 交互的库。此时，您可能会对像 Spark 这样的分布式引擎如何利用 Python 作为其 API 感到好奇。这不是一件容易的事——基于 JVM 的语言（例如 Java 和 Scala）与 Python 有着根本的不同。Java 是一种编译型语言，这意味着一切都被翻译成机器码，这样它就可以在运行时由处理器执行。另一方面，Python 是一种解释型语言。1正如预期的那样，编译程序比解释程序运行得更快，因为它们可以直接由 CPU 执行。

本章将涵盖以下领域：

• Apache Spark 架构

• Spark 基础知识

• DataFrames 不变性

• PySpark 如何工作并与 Spark JVM 通信

• 为什么 pandas 与 Spark 不同

• Scikit-learn 与 PySpark 的机器学习功能对比

Spark架构

Spark 架构由以下主要组件组成：驱动程序、工作程序、集群管理器和执行程序。每一个对于 Spark 分布式系统的工作方式都是必不可少的。让我们将图 2-1分解成它的组成部分：

Driver Program

驱动程序（又名 Spark Driver）是在驱动程序机器上运行的专用进程。它负责执行和持有封装了SparkContext的Spark Session。它被认为是应用程序的入口或“真正的程序”。它持有的上下文具有所有基本功能、启动时交付的上下文以及有关集群的信息。该驱动程序还包含有向无环图 (DAG) 调度程序、任务调度程序、块管理器，以及将代码转换为工作程序和执行程序可以在集群上执行的作业所需的一切。驱动程序与集群管理器协同工作，以查找现有机器和分配的资源。

Worker Node

顾名思义，工作节点负责执行工作。它本身包含执行程序。工作节点有专用的共享内存，可以让多个执行器运行，并可以服务于多个 Spark 应用程序。

Executor

执行器是为工作节点上的 Spark 应用程序启动的进程。多个任务可以分配给每个执行者。JVM 进程接收任务并与集群管理器通信。同一个执行器上的任务可以受益于共享内存，例如缓存和全局参数，这使得任务运行速度很快。

Task

任务是最小的工作单元，运行分配给它的所需代码，以及分配给它的数据片段。

Cluster Manager

与驱动程序一起，集群管理器负责在分布式系统的管弦乐队中演奏：将工作人员和执行程序分配给任务并将资源可用性传达给驱动程序。实际上，它可以是 Kubernetes 2、mesos 3、Hadoop Yarn、4或任何其他可以管理机器和网络容量的集群管理器。

图 2-1。- Spark driver-worker 架构

所讨论的每个组件在大规模编排 Spark 程序中都起着关键作用。Spark 驱动程序可以启动两个作业并将工作分配给令人兴奋的执行程序。执行者之间的唯一区别是他们对专用存储的访问。如图 2-2所示，每个人都获得一个独特的存储范围来读取和运行他们的计算。

图 2-2。- Spark 推出两项工作

此时，您已经知道 Python 是一种不同于 JVM 家族中所有其他语言的语言。您还知道，Spark 的核心是运行基于 JVM 的进程。让我们看看 Spark 如何与 Python 一起工作。

在最基本的层面上，组件可以通过 API 和共享网络进行通信。这意味着如果我有一个运行 JVM 进程的任务，它可以利用进程间通信。IPC 与 Python 一起工作。

我写了一个 PySpark 应用程序。它如何与 Spark 一起工作？

本质上，每次启动 PySpark 作业时，它都会在后台创建两个应用程序：Python 和 JVM。

Python 是定义代码的主程序，而 JVM 是进行 Spark 查询优化、计算、将任务分配到集群等的程序。在 PySpark 应用程序中，SparkContext 本身有一个名为 _gateway 的参数，它负责保存将 Py4J 应用程序传递给 JVM Spark 服务器的上下文。

等等，什么是 Py4J 应用程序？

根据其文档，Py4J 使在 Python 解释器中运行的 Python 程序能够动态访问 Java 虚拟机 (JVM) 中的 Java 对象。调用方法就像 Java 对象驻留在 Python 解释器中一样，并且可以通过标准 Python 集合方法访问 Java 集合。

在我们的例子中，这意味着 Python 代码可以与 JVM 通信以传递代码要求和对象。图 2-3显示了它是如何工作的。当 PySpark 驱动程序启动时，它会启动一个 Spark JVM 应用程序，其中 Py4J 服务器配置为直接与 Spark JVM 通信。信息在 PySpark 驱动程序和 Py4J 之间传输，在 Python 中通常称为 pickled items。5个

图 2-3。- Py4J 服务充当 Python 应用程序和 Spark JVM 之间的中介

我们还没有完成。由于 Python 是一个解释器并且代码没有被高级编译，因此有一个执行器在每个工作节点中保存 Python 代码。执行程序会在需要执行逻辑时启动 Python 应用程序。这里有一些例外和多年来添加的优化。粗略地说，众所周知，PySpark 在运行时的效率通常低于传统的 Scala/Java 代码。

Apache Spark 基础知识

Spark 的基础知识从分布式系统开始，并涉及软件架构以及围绕调度和计算的决策。

软件架构

由于 Spark 是作为支持各种分布式计算工作负载的通用引擎构建的，因此其软件架构是分层构建的，如图 2-4 所示。

底层使用RDD（弹性分布式数据集API 和数据源连接器）将存储抽象出来。Spark 中的存储可以是任何具有可读格式的东西（更多内容在第 4 章）。

顶层包含供我们使用的 API 和库。它通过在 Spark DataFrame或 Dataset API上执行来抽象出所有 Spark 内部结构。DataFrame 是按列组织的分布式数据集合。您可以将其想象成数据库中的一张表。它有专门的结构和格式，你可以在上面运行特定的操作。它还带有一个模式，其中每一列都支持特定的数据类型：整数、长字符串、数组、映射、字符、日期、时间戳、二进制等.). Datasets 和 DataFrame 之间的主要区别在于 Datasets 带有列的类型安全，这意味着我们无法将字符串类型的列误认为是 int。然而，我们为此付出了代价。在 Datasets 上操作被认为比在 DataFrame 上操作慢。

请注意，架构可以是您正在读入 DataFrame 的数据的一部分。尽管 Spark 尽力从 CSV 文件推断模式，但很多时候，由于特定的 UTF 格式，它可能会推断出错误的模式。时不时地，分隔符会被错误地结束，或者整个 CSV 使用其他格式保存。

要将 CSV 读入 DataFrame，您只需使用专用格式函数调用读取操作：

df = spark.read.csv("some files")

按照上面的代码片段df.printSchema()将执行读取功能并打印模式供您检查。如果您愿意，您还可以使用选项功能更改分隔符 - 使用选项将配置传递给阅读器.options(delimiter=',').

用户特定的自定义架构

通常，您会想要控制跟随并提供您的自定义模式。这使得代码本身的协作和可重复性成为可能。它还可以节省您以后调试问题的宝贵时间。

那么，如何使用 Spark 做到这一点？

您将需要创建一个 StructType() 并在阅读期间将其作为所需模式传递给阅读器。在结构类型中，使用专用 API 添加所有列名称和类型。例如，如果我在 Python 中有一个 int 类型的列，在 PySpark 中，我将启动 DataType.integerType，如以下代码示例中所定义：

schema = StructType() \
      .add("Number",IntegerType(),True) \
      .add("City",StringType(),True) \
      .add("Age",DoubleType(),True) \
      .add("Decommisioned",BooleanType(),True))
      
df_with_schema = spark.read.format("csv") \
      .option("header", True) \
      .schema(schema) \
      .load("{some_file_path}.csv")

在上面的示例中，添加功能中有 True。这意味着该值可以为空。

提示

为了代码的可重用性和模块化，将架构写入 JSON 文件。这将使您能够快速加载文件并将架构配置与代码本身分离。这也是一种更简洁的代码编写方式。

您需要做的就是调用 df_with_schema.schema.json()

它将输出以下内容：

[json,code]
{
   "type" : "struct",
   "fields" : [  {
     "name" : "Number",
     "type" : "integer",
     "nullable" : true,
     "metadata" : { }
   }, {
     "name" : "City",
     "type" : "string",
     "nullable" : true,
     "metadata" : { }
   }, {
     "name" : "Age",
     "type" : "float",
     "nullable" : true,
     "metadata" : { }
   }, {
     "name" : "Decommisioned",
     "type" : "bool",
     "nullable" : true,
     "metadata" : { }
   } ]
 }

将此输出保存到您选择的文件中，只需使用带有函数的 Python：

with open(schema.json, 'w') as f:
    sys.stdout = f # Change the standard output to the file
    print(df_with_schema.schema.json()) 

现在使用 fromJson 功能将其加载回来，您可以在本示例中执行以下操作：

import json
schemaFromJson = StructType.fromJson(json.loads(schema.json))

现在您有了模式的实例，您可以按照前面讨论的那样读取 DataFrame。

如何将 Python 数据类型转换为 Spark？

或者更好的是，PySpark 如何解释类型？看一下表 2-1，它将作为您需要为数据提供模式或执行时的情况的指南：

表 2-1。基本 Python 数据类型以及如何启动它们以设计模式。

请注意，虽然 Python 中有许多类型，但在启动模式定义的类型时，有些类型会重复出现 - 如 Long 和 Integer。在 Python 的赋值过程中，它们的值都将转换为 int。PySpark 提出的这一挑战源于使用 JVM 和 Python。

RDD抽象了什么？

这是一个非常棒的问题，它深入探讨了分布式存储和数据源连接器的概念。Spark 在这里隐藏了许多复杂性。RDD 结合了应用程序依赖性、分区和 -> 计算功能。

基本上，您可以将分区视为 分布式数据。它们是数据块。分区是在一个或多个区域中管理的数据的一部分。它是数据本身的实际逻辑划分。RDD 的目的是将分区连接到逻辑迭代器，给定执行程序可以迭代的应用程序依赖项。分区非常重要，因为它们为 Spark 提供了在执行程序之间轻松拆分工作的能力。

DataFrames/Datasets 和 RDDs API 之间有什么区别？

RDD 本质上是属于下层的 JVM 对象的分布式集合，而 DataFrames/Datasets 类似，后者受益于 Spark Catalyst 查询优化。

图 2-4。- spark软件架构

Spark 催化剂是 Spark 优势的最重要组成部分之一。通过创建操作树并根据明确定义的规则对其进行修剪，查询和执行得到优化。因此，Spark 运行速度更快，成本更低。只要有可能，建议利用顶层 API 并避免 RDD。

Catalyst 中有很多创新，深入这个世界真的很有趣。但是 - 鉴于它对 ML 工作负载几乎没有影响 - 我不会深入研究它。

Spark 使我们能够与 PySpark 交互的 API 如下：

ML

大规模运行和应用机器学习工作负载。

GraphX/GraphFrames

图上的处理逻辑。该组件支持在有边和节点的图形数据上运行图形操作。GraphX 优化了顶点和边类型的表示，当它们是原始数据类型（例如，int、double 等）时，通过将它们存储在专用数组中来减少内存占用。虽然机器学习与图形计算有很大关系，但这两个库是分开的，用于不同的目的。但是，它确实有一些有趣的算法，可以帮助您丰富数据并运行预处理和特征工程。例如，PageRank 算法在 GraphX 中可用。当查看带有顶点和边的图时，它会根据顶点的重要性对顶点进行排序。想想推特。如果一个用户被很多人关注，他们的排名就会很高。

TensorFrames

一个不幸被弃用的图书馆。它的目的是充当 TensorFlow 的包装器以与 Spark DataFrames 一起工作。

Structured Streaming

此 API 启动一个永无止境的 Spark 作业，以流式处理微批处理数据。还有一个改进的引擎，用于降低延迟，称为连续处理。Spark 应用程序本身会创建一个侦听器，等待收集新数据。自从这个到来后，Spark SQL 负责处理并更新最终结果。

DataFrame 是不可变的

重要的是要注意 DataFrame、数据集和 RDD 被认为是不可变的存储。不变性意味着对象的状态在创建后不能更改。反过来，这意味着在编写 PySpark 代码时，我们必须牢记这一点。每次我们在 DataFrame 上执行时，我们可能会创建一个新的 DataFrame。本质上，Spark 本身在幕后优化了许多事情，但记住这一点至关重要。

让我们看一下代码示例，我们在其中读取 DataFrame 并对其运行选择操作：

train_df = spark.read.csv(‘training_data.csv’, header = True)
train_df.select(‘bot’)

仔细查看上面的代码片段。这里我们读取了一个DataFrame，并将DataFrame赋值给train_df参数。

选择操作后 train_df 会发生什么？没有！是的，绝对没有。这就是不变性的力量：对象本身不会改变。本质上，train_df 现在指向表示 DataFrame 的内存。表示本身不能改变。发生的事情是我们运行选择操作以仅选择机器人数据并且没有捕获新的潜在指针。因此，DataFrame 是相同的，并且 train_df 仍然指向内存中表示 DataFrame 的相同位置。

比那更多的。Spark 有惰性执行。

惰性执行 - Spark 不会开始执行流程，直到调用 Action 类型的操作。Action 可以是任何返回不属于 DataFrame 的值的东西。例如，DataFrame 中的行数或写入操作。

回到我们的例子，Spark 甚至没有将数据加载到内存中。它创建了一个操作图，其中包含读取和选择。之后，如果没有任何带有select的DataFrame操作，Spark会对其进行剪枝，根本不会执行。这就是 Spark 的惰性执行和不变性保护我们不犯错误的地方。请注意它。

所以，我们能做些什么？我们可以将结果 DataFrame 分配到一个实例中，如代码示例所示：

train_df = spark.read.csv(‘training_data.csv’, header = True)
tmp_df = train_df.select(‘bot’)

此处，代码片段保存了对 tmp_df 的引用。如果我们稍后继续对其执行并有一个动作操作，Spark 会将其作为其图形的一部分执行。

您可能会问自己：为什么不变性很重要？

操作的不变性是函数式编程的支柱之一。函数式编程旨在构建可靠的应用程序。这里的目标是对象在创建后不应更改其状态。这样，开发人员将能够跟踪对对象所做的更改，类似于事件链。

在使用 Spark 时，不变性可以更轻松地了解发生了哪些操作并对其进行跟踪。因此，从本质上讲，每个 DataFrame 都是特定操作的结果，该操作是可重现和可跟踪的。这很关键，在分布式系统群中变得更具挑战性。因此，选择不可变确实使我们能够在操作中实现弹性。不过，为了节省内存，我们无法保存生成的每个数据帧。考虑到这一点并学习以一种能够重现机器学习实验的方式使用 Spark（第 3 章会详细介绍）。

PySpark 和函数式编程

我之前提到过函数式编程，这并非偶然。Spark 从函数式编程中借用了许多概念，从匿名函数开始。

匿名无状态函数

在没有状态的情况下执行的函数尚未命名。在数学中，它们通常被称为 lambda 演算。如果在阅读本书或就 Spark 进行对话时，您听到“lambda 函数”，您就会知道它的来源。

本质上，您可以将匿名函数传递给 RDD 以在数据上执行它。看一下代码片段：

rdd2=rdd.map(lambda x: (x,1))

这个代码片段有一个 RDD 实例，它调用了一个映射功能。在功能映射内部，它调用匿名函数。对于每个 x - 意思是，RDD 中的每一行都变成一对。你可以把它想象成。rdd2 现在拥有图形操作，要求将每个 x 变成 (x,1)。

该映射利用传递可重复且独立于状态的独立函数的功能。

能够传递无状态函数以在 RDD 上执行使得并行执行变得毫不费力。想一想：每个执行者都获取一个数据片段并运行映射并返回一个答案。无需交换信息，这使得该系统具有极强的可扩展性和弹性。如果一个节点出现故障，则无需重新计算所有内容 - 只需重新计算它负责的那部分数据。最终结果与在单个节点上执行时一样。

举一个更高级的例子，让我们假设有一个请求，要求从所有值中找出最大值。现在需要研究具有最大操作的减少操作。看下一段代码：

rdd3=rdd2.reduce((a,b)=> (("max",a._2 max b._2)._2))

我们这里使用了reduce。Reduce 将计算所有值中的最大值，从每对中的局部值开始。在本地值之后，数据将被洗牌（移动）到一个专用的执行器，该执行器将计算最终结果。在我们的例子中，将返回 3。

尽管函数式编程有更多的原则，但 Spark 并不一定尽可能地遵循它们。主要有：

• 不变性，

• 纪律严明的国家——最大限度地减少对国家的依赖。

执行 PySpark 代码涉及哪些步骤？

在配置集群的地方运行 PySpark 代码：在云中，您需要配置机器并提供权限。执行 Python 代码本身需要什么？它与 JVM 进程非常相似。如前所述，PySpark 代码将处理其余部分。

要使用专用笔记本在本地执行，请按照Github 存储库中的说明进行操作。

Spark 也有 Shell 命令：PySpark shell 是一个交互式 shell，使您能够使用命令行尝试 PySpark。PySpark shell 负责将 Python API 链接到 Spark 核心并初始化 SparkContext。它基于名为 REPL（Read-Eval-Print-Loop）的 Scala 概念。要执行书中的练习，GitHub 存储库中的说明应该适合您。对于云安装，如果您不是 IT 专业人员，最好利用托管 Spark 解决方案或咨询 IT。此外，您还可以查看 Spark 文档quickstart。

pandas DataFrame 与 Spark DataFrame

Spark DataFrames 曾经受到pandas的启发，这使得一开始很容易混淆两者。pandas 是一个被广泛采用的用于数据操作和分析的库。许多开发人员使用它来使用 Python 推断数据。它还在称为 DataFrame 的数据之上提供了一个抽象。阅读 pandas DataFrame 很简单——下面是一个代码示例，展示了如何做到这一点：

import pandas as pd
df = pd.read_csv(....)

但是，其核心存在许多差异。大熊猫不是为规模而建的。它旨在对适合一台机器内存的数据进行操作。它没有我们在本章开头讨论的分布式 Spark 架构。此外，它也不遵循函数式编程原则。DataFrame 本身是可变的。

表 2-2。Spark DataFrame 与 Pandas DataFrame

在表 2-2 中，您可以看到没有现成的方法可以在 panada 的 DataFrame 上并行操作。这并不意味着它完全不可能。它只是意味着您必须创建一个解决方案并考虑可能的罪魁祸首（线程锁、竞争条件等）及其对最终结果的影响。

同样，Spark 支持惰性操作，而在 pandas 中，操作会在 Python 行执行时立即发生。最重要的是，数据本身在 pandas 中会发生变化，这意味着它不是不可变的。这使得在 pandas 上记忆和操作变得容易。然而，这也使得使用并行或分布式计算进行扩展具有挑战性。

Scikit-learn 与 Pyspark ML

Scikit-learn (sklearn) 和 Pyspark ML 有时也会混淆。Scikit-learn 是一个用 Python 编写的机器学习库，它利用了著名的 Python 库，例如 numPy、SciPy 等。虽然众所周知，它在处理适合 RAM 的数据时具有出色的性能，但它以非分布式方式运行缓慢。Spark 增加了配置集群以在更大规模上运行的开销，以及针对并行/分布式执行调整的算法。

什么时候使用哪种工具？

• 最好使用 Spark 来处理您的大型数据集（GB-TB 甚至 PB 规模），并使用 MLLib/ML 对其执行机器学习。

• 当您的所有数据集都适合您机器的内存时，可以使用 Scikit-learn 以及 pandas 和其他库在 Python 上构建您的算法。

sklean 使用数据集的概念，类似于以前的 Python 库，例如 pandas。这里的数据集是可变的，开箱即用的扩展受到限制，并且依赖于与其他工具的结合。此外，模型部署类似，可以保存到磁盘并使用多个 API 重新加载。

当您到达讨论 Spark ML Pipelines 的第 6 章时，相似性将变得更加生动。这是因为管道的整个概念都受到了 Scikit-learn 项目和社区的启发。Spark 社区主动决定在这里使用类似的概念，并在设计 DataFrame 时尽可能地缓解和拉平 Spark 的学习曲线。

表 2-3。Scikit Learn 与 Spark ML

概括

在这一点上，我希望您已经准备好并准备好全面踏上未来的旅程，并获得 Spark 和机器学习管道的实践经验。使用 Spark 进行端到端机器学习管道的转变需要专门的技术和对细节的关注。本章涵盖了 Spark 的基础知识，让您深入了解其架构和 API，甚至将其与其他受欢迎的 Python 工具进行了比较。要记住一件事：机器学习项目是长期的，需要时间、努力和协作。请记住保持代码可重用、友好和模块化。这将使您能够跟踪工作并跟踪本书的实践练习。文章来源地址https://www.toymoban.com/news/detail-408038.html