Spark 4.0.0 支持 Python 3.9+。它可以使用标准的 CPython 解释器，因此可以使用 NumPy 等 C 库。它也支持 PyPy 7.3.6+。

Python 中的 Spark 应用程序可以通过 bin/spark-submit 脚本（运行时包含 Spark）运行，或者通过将其包含在 setup.py 中来运行，如下所示：

    install_requires=[
        'pyspark==4.0.0'
    ]

要在不通过 pip 安装 PySpark 的情况下运行 Python Spark 应用程序，请使用 Spark 目录中的 bin/spark-submit 脚本。此脚本将加载 Spark 的 Java/Scala 库，并允许您向集群提交应用程序。您也可以使用 bin/pyspark 启动交互式 Python Shell。

如果您希望访问 HDFS 数据，您需要使用链接到您的 HDFS 版本的 PySpark 构建。Spark 主页上也提供了常见 HDFS 版本的预构建包。

最后，您需要将一些 Spark 类导入到您的程序中。添加以下行：

from pyspark import SparkContext, SparkConf

PySpark 要求驱动程序和工作节点使用相同的小版本 Python。它使用 PATH 中的默认 Python 版本，您可以通过 PYSPARK_PYTHON 指定要使用的 Python 版本，例如：

$ PYSPARK_PYTHON=python3.8 bin/pyspark
$ PYSPARK_PYTHON=/path-to-your-pypy/pypy bin/spark-submit examples/src/main/python/pi.py

Spark 4.0.0 默认构建并分发以支持 Scala 2.13。（Spark 也可以构建以支持其他 Scala 版本。）要用 Scala 编写应用程序，您需要使用兼容的 Scala 版本（例如 2.13.X）。

要编写 Spark 应用程序，您需要添加对 Spark 的 Maven 依赖。Spark 可通过 Maven Central 获取，地址为：

groupId = org.apache.spark
artifactId = spark-core_2.13
version = 4.0.0

此外，如果您希望访问 HDFS 集群，您需要为您的 HDFS 版本添加 hadoop-client 依赖。

groupId = org.apache.hadoop
artifactId = hadoop-client
version = <your-hdfs-version>

最后，您需要将一些 Spark 类导入到您的程序中。添加以下行：

import org.apache.spark.SparkContext
import org.apache.spark.SparkConf

(在 Spark 1.3.0 之前，您需要显式 import org.apache.spark.SparkContext._ 来启用必要的隐式转换。)

Spark 4.0.0 支持lambda 表达式来简洁地编写函数，否则您可以使用 org.apache.spark.api.java.function 包中的类。

请注意，Spark 2.2.0 中已移除对 Java 7 的支持。

要在 Java 中编写 Spark 应用程序，您需要添加对 Spark 的依赖。Spark 可通过 Maven Central 获取，地址为：

groupId = org.apache.spark
artifactId = spark-core_2.13
version = 4.0.0

此外，如果您希望访问 HDFS 集群，您需要为您的 HDFS 版本添加 hadoop-client 依赖。

groupId = org.apache.hadoop
artifactId = hadoop-client
version = <your-hdfs-version>

最后，您需要将一些 Spark 类导入到您的程序中。添加以下行：

import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.SparkConf;

Spark 程序首先必须创建一个 SparkContext 对象，它告诉 Spark 如何访问集群。要创建 SparkContext，您首先需要构建一个 SparkConf 对象，其中包含有关您应用程序的信息。

conf = SparkConf().setAppName(appName).setMaster(master)
sc = SparkContext(conf=conf)

Spark 程序首先必须创建一个 SparkContext 对象，它告诉 Spark 如何访问集群。要创建 SparkContext，您首先需要构建一个 SparkConf 对象，其中包含有关您应用程序的信息。

每个 JVM 只能有一个 SparkContext 处于活动状态。在创建新的 SparkContext 之前，您必须先 stop() 当前活动的 SparkContext。

val conf = new SparkConf().setAppName(appName).setMaster(master)
new SparkContext(conf)

Spark 程序首先必须创建一个 JavaSparkContext 对象，它告诉 Spark 如何访问集群。要创建 SparkContext，您首先需要构建一个 SparkConf 对象，其中包含有关您应用程序的信息。

SparkConf conf = new SparkConf().setAppName(appName).setMaster(master);
JavaSparkContext sc = new JavaSparkContext(conf);

在 PySpark Shell 中，一个特殊的、解释器感知的 SparkContext 已经为您创建，位于名为 sc 的变量中。创建您自己的 SparkContext 将不起作用。您可以使用 --master 参数设置上下文连接的 master，并且可以通过向 --py-files 传递逗号分隔的列表来将 Python .zip、.egg 或 .py 文件添加到运行时路径。有关第三方 Python 依赖项，请参阅Python 包管理。您还可以通过向 --packages 参数提供逗号分隔的 Maven 坐标列表，将依赖项（例如 Spark Packages）添加到您的 Shell 会话中。任何可能存在依赖项的其他仓库（例如 Sonatype）都可以传递给 --repositories 参数。例如，要在恰好四个核心上运行 bin/pyspark，请使用：

$ ./bin/pyspark --master "local[4]"

或者，要将 code.py 也添加到搜索路径（以便稍后能够 import code），请使用：

$ ./bin/pyspark --master "local[4]" --py-files code.py

有关选项的完整列表，请运行 pyspark --help。在幕后，pyspark 调用了更通用的spark-submit 脚本。

也可以在 IPython（增强型 Python 解释器）中启动 PySpark Shell。PySpark 支持 IPython 1.0.0 及更高版本。要使用 IPython，在运行 bin/pyspark 时将 PYSPARK_DRIVER_PYTHON 变量设置为 ipython：

$ PYSPARK_DRIVER_PYTHON=ipython ./bin/pyspark

要使用 Jupyter Notebook（以前称为 IPython Notebook），

$ PYSPARK_DRIVER_PYTHON=jupyter PYSPARK_DRIVER_PYTHON_OPTS=notebook ./bin/pyspark

您可以通过设置 PYSPARK_DRIVER_PYTHON_OPTS 来自定义 ipython 或 jupyter 命令。

Jupyter Notebook 服务器启动后，您可以从“文件”选项卡创建新笔记本。在笔记本内部，您可以在开始从 Jupyter Notebook 尝试 Spark 之前，将命令 %pylab inline 作为笔记本的一部分输入。

在 Spark Shell 中，一个特殊的、解释器感知的 SparkContext 已经为您创建，位于名为 sc 的变量中。创建您自己的 SparkContext 将不起作用。您可以使用 --master 参数设置上下文连接的 master，并且可以通过向 --jars 参数传递逗号分隔的列表来将 JAR 添加到 classpath。您还可以通过向 --packages 参数提供逗号分隔的 Maven 坐标列表，将依赖项（例如 Spark Packages）添加到您的 Shell 会话中。任何可能存在依赖项的其他仓库（例如 Sonatype）都可以传递给 --repositories 参数。例如，要在恰好四个核心上运行 bin/spark-shell，请使用：

$ ./bin/spark-shell --master "local[4]"

或者，要将 code.jar 也添加到其 classpath，请使用：

$ ./bin/spark-shell --master "local[4]" --jars code.jar

要使用 Maven 坐标包含依赖项：

$ ./bin/spark-shell --master "local[4]" --packages "org.example:example:0.1"

有关选项的完整列表，请运行 spark-shell --help。在幕后，spark-shell 调用了更通用的spark-submit 脚本。

通过在驱动程序中对现有可迭代对象或集合调用 SparkContext 的 parallelize 方法来创建并行化集合。集合中的元素被复制以形成一个可以并行操作的分布式数据集。例如，以下是如何创建一个包含数字 1 到 5 的并行化集合：

data = [1, 2, 3, 4, 5]
distData = sc.parallelize(data)

创建后，分布式数据集 (distData) 可以并行操作。例如，我们可以调用 distData.reduce(lambda a, b: a + b) 来将列表元素相加。我们稍后会介绍分布式数据集上的操作。

通过在驱动程序中对现有集合（一个 Scala Seq）调用 SparkContext 的 parallelize 方法来创建并行化集合。集合中的元素被复制以形成一个可以并行操作的分布式数据集。例如，以下是如何创建一个包含数字 1 到 5 的并行化集合：

val data = Array(1, 2, 3, 4, 5)
val distData = sc.parallelize(data)

创建后，分布式数据集 (distData) 可以并行操作。例如，我们可以调用 distData.reduce((a, b) => a + b) 来将数组元素相加。我们稍后会介绍分布式数据集上的操作。

通过在驱动程序中对现有 Collection 调用 JavaSparkContext 的 parallelize 方法来创建并行化集合。集合中的元素被复制以形成一个可以并行操作的分布式数据集。例如，以下是如何创建一个包含数字 1 到 5 的并行化集合：

List<Integer> data = Arrays.asList(1, 2, 3, 4, 5);
JavaRDD<Integer> distData = sc.parallelize(data);

创建后，分布式数据集 (distData) 可以并行操作。例如，我们可以调用 distData.reduce((a, b) -> a + b) 来将列表元素相加。我们稍后会介绍分布式数据集上的操作。

PySpark 可以从 Hadoop 支持的任何存储源创建分布式数据集，包括本地文件系统、HDFS、Cassandra、HBase、Amazon S3 等。Spark 支持文本文件、SequenceFiles 和任何其他 Hadoop InputFormat。

文本文件 RDD 可以使用 SparkContext 的 textFile 方法创建。此方法接受文件的 URI（可以是机器上的本地路径，也可以是 hdfs://、s3a:// 等 URI），并将其作为行集合读取。以下是调用示例：

>>> distFile = sc.textFile("data.txt")

创建后，distFile 可以通过数据集操作进行处理。例如，我们可以使用 map 和 reduce 操作将所有行的长度相加，如下所示：distFile.map(lambda s: len(s)).reduce(lambda a, b: a + b)。

Spark 读取文件的一些注意事项

• 如果使用本地文件系统中的路径，则该文件也必须在工作节点上的相同路径可访问。要么将文件复制到所有工作节点，要么使用网络挂载的共享文件系统。

• Spark 所有基于文件的输入方法，包括 textFile，都支持对目录、压缩文件和通配符进行操作。例如，您可以使用 textFile("/my/directory")、textFile("/my/directory/*.txt") 和 textFile("/my/directory/*.gz")。

• textFile 方法还接受一个可选的第二个参数，用于控制文件的分区数量。默认情况下，Spark 为文件的每个块创建一个分区（HDFS 中块的默认大小为 128MB），但您也可以通过传递更大的值来请求更多的分区。请注意，分区数量不能少于块的数量。

除了文本文件，Spark 的 Python API 还支持其他几种数据格式：

• SparkContext.wholeTextFiles 允许您读取包含多个小文本文件的目录，并将它们作为 (filename, content) 对返回。这与 textFile 不同，textFile 会为每个文件中的每行返回一条记录。

• RDD.saveAsPickleFile 和 SparkContext.pickleFile 支持以一种简单的格式保存 RDD，该格式由 pickled Python 对象组成。pickle 序列化使用批处理，默认批处理大小为 10。

• SequenceFile 和 Hadoop 输入/输出格式

注意：此功能目前标记为 Experimental（实验性），旨在供高级用户使用。将来可能会被基于 Spark SQL 的读写支持所取代，届时 Spark SQL 将是首选方法。

Writable 支持

PySpark SequenceFile 支持在 Java 内部加载键值对的 RDD，将 Writable 类型转换为基本的 Java 类型，然后使用 pickle 序列化生成的 Java 对象。当将键值对的 RDD 保存到 SequenceFile 时，PySpark 执行相反的操作：它将 Python 对象反序列化为 Java 对象，然后将它们转换为 Writable 类型。以下 Writable 类型会自动转换：

数组不能开箱即用。用户在读写时需要指定自定义的 ArrayWritable 子类型。在写入时，用户还需要指定自定义转换器，将数组转换为自定义的 ArrayWritable 子类型。在读取时，默认转换器会将自定义的 ArrayWritable 子类型转换为 Java Object[]，然后这些对象会被 pickled 为 Python 元组。要获取原始类型数组的 Python array.array，用户需要指定自定义转换器。

保存和加载 SequenceFiles

与文本文件类似，可以通过指定路径来保存和加载 SequenceFiles。可以指定键和值类，但对于标准 Writable 类型则不需要。

>>> rdd = sc.parallelize(range(1, 4)).map(lambda x: (x, "a" * x))
>>> rdd.saveAsSequenceFile("path/to/file")
>>> sorted(sc.sequenceFile("path/to/file").collect())
[(1, u'a'), (2, u'aa'), (3, u'aaa')]

保存和加载其他 Hadoop 输入/输出格式

PySpark 还可以读取任何 Hadoop InputFormat 或写入任何 Hadoop OutputFormat，支持“新”和“旧”的 Hadoop MapReduce API。如果需要，可以将 Hadoop 配置作为 Python 字典传入。以下是使用 Elasticsearch ESInputFormat 的示例：

$ ./bin/pyspark --jars /path/to/elasticsearch-hadoop.jar
>>> conf = {"es.resource" : "index/type"}  # assume Elasticsearch is running on localhost defaults
>>> rdd = sc.newAPIHadoopRDD("org.elasticsearch.hadoop.mr.EsInputFormat",
                             "org.apache.hadoop.io.NullWritable",
                             "org.elasticsearch.hadoop.mr.LinkedMapWritable",
                             conf=conf)
>>> rdd.first()  # the result is a MapWritable that is converted to a Python dict
(u'Elasticsearch ID',
 {u'field1': True,
  u'field2': u'Some Text',
  u'field3': 12345})

请注意，如果 InputFormat 仅依赖于 Hadoop 配置和/或输入路径，并且键和值类可以根据上述表格轻松转换，则此方法适用于此类情况。

如果您有自定义的序列化二进制数据（例如从 Cassandra / HBase 加载数据），那么您首先需要将这些数据在 Scala/Java 端转换为可以由 pickle 的 pickler 处理的格式。为此提供了一个 Converter trait。只需扩展此 trait 并在 convert 方法中实现您的转换代码即可。请记住，确保此类别以及访问您的 InputFormat 所需的任何依赖项都打包到您的 Spark 作业 jar 中，并包含在 PySpark classpath 中。

有关使用自定义转换器使用 Cassandra / HBase InputFormat 和 OutputFormat 的示例，请参阅Python 示例和Converter 示例。

Spark 可以从 Hadoop 支持的任何存储源创建分布式数据集，包括本地文件系统、HDFS、Cassandra、HBase、Amazon S3 等。Spark 支持文本文件、SequenceFiles 和任何其他 Hadoop InputFormat。

文本文件 RDD 可以使用 SparkContext 的 textFile 方法创建。此方法接受文件的 URI（可以是机器上的本地路径，也可以是 hdfs://、s3a:// 等 URI），并将其作为行集合读取。以下是调用示例：

scala> val distFile = sc.textFile("data.txt")
distFile: org.apache.spark.rdd.RDD[String] = data.txt MapPartitionsRDD[10] at textFile at <console>:26

创建后，distFile 可以通过数据集操作进行处理。例如，我们可以使用 map 和 reduce 操作将所有行的长度相加，如下所示：distFile.map(s => s.length).reduce((a, b) => a + b)。

Spark 读取文件的一些注意事项

• 如果使用本地文件系统中的路径，则该文件也必须在工作节点上的相同路径可访问。要么将文件复制到所有工作节点，要么使用网络挂载的共享文件系统。

• Spark 所有基于文件的输入方法，包括 textFile，都支持对目录、压缩文件和通配符进行操作。例如，您可以使用 textFile("/my/directory")、textFile("/my/directory/*.txt") 和 textFile("/my/directory/*.gz")。当读取多个文件时，分区的顺序取决于文件系统返回文件的顺序。它可能（也可能不）遵循文件按路径的字典顺序。在一个分区内，元素的顺序是根据它们在底层文件中的顺序排列的。

• textFile 方法还接受一个可选的第二个参数，用于控制文件的分区数量。默认情况下，Spark 为文件的每个块创建一个分区（HDFS 中块的默认大小为 128MB），但您也可以通过传递更大的值来请求更多的分区。请注意，分区数量不能少于块的数量。

除了文本文件，Spark 的 Scala API 还支持其他几种数据格式：

• SparkContext.wholeTextFiles 允许您读取包含多个小文本文件的目录，并将它们作为 (filename, content) 对返回。这与 textFile 不同，textFile 会为每个文件中的每行返回一条记录。分区由数据局部性决定，在某些情况下可能会导致分区过少。对于这些情况，wholeTextFiles 提供了一个可选的第二个参数，用于控制最小分区数量。

• 对于 SequenceFiles，使用 SparkContext 的 sequenceFile[K, V] 方法，其中 K 和 V 是文件中键和值的类型。这些类型应该是 Hadoop Writable 接口的子类，例如 IntWritable 和 Text。此外，Spark 允许您为一些常见的 Writable 指定原生类型；例如，sequenceFile[Int, String] 将自动读取 IntWritables 和 Texts。

• 对于其他 Hadoop InputFormats，您可以使用 SparkContext.hadoopRDD 方法，该方法接受任意的 JobConf、输入格式类、键类和值类。设置这些参数的方式与您在 Hadoop 作业中使用输入源的方式相同。您也可以使用 SparkContext.newAPIHadoopRDD 处理基于“新”MapReduce API (org.apache.hadoop.mapreduce) 的 InputFormats。

• RDD.saveAsObjectFile 和 SparkContext.objectFile 支持以一种由序列化 Java 对象组成的简单格式保存 RDD。虽然这不如 Avro 等专用格式高效，但它提供了一种保存任何 RDD 的简单方法。

Spark 可以从 Hadoop 支持的任何存储源创建分布式数据集，包括本地文件系统、HDFS、Cassandra、HBase、Amazon S3 等。Spark 支持文本文件、SequenceFiles 和任何其他 Hadoop InputFormat。

文本文件 RDD 可以使用 SparkContext 的 textFile 方法创建。此方法接受文件的 URI（可以是机器上的本地路径，也可以是 hdfs://、s3a:// 等 URI），并将其作为行集合读取。以下是调用示例：

JavaRDD<String> distFile = sc.textFile("data.txt");

创建后，distFile 可以通过数据集操作进行处理。例如，我们可以使用 map 和 reduce 操作将所有行的长度相加，如下所示：distFile.map(s -> s.length()).reduce((a, b) -> a + b)。

Spark 读取文件的一些注意事项

• 如果使用本地文件系统中的路径，则该文件也必须在工作节点上的相同路径可访问。要么将文件复制到所有工作节点，要么使用网络挂载的共享文件系统。

• Spark 所有基于文件的输入方法，包括 textFile，都支持对目录、压缩文件和通配符进行操作。例如，您可以使用 textFile("/my/directory")、textFile("/my/directory/*.txt") 和 textFile("/my/directory/*.gz")。

• textFile 方法还接受一个可选的第二个参数，用于控制文件的分区数量。默认情况下，Spark 为文件的每个块创建一个分区（HDFS 中块的默认大小为 128MB），但您也可以通过传递更大的值来请求更多的分区。请注意，分区数量不能少于块的数量。

除了文本文件，Spark 的 Java API 还支持其他几种数据格式：

• JavaSparkContext.wholeTextFiles 允许您读取包含多个小文本文件的目录，并将它们作为 (filename, content) 对返回。这与 textFile 不同，textFile 会为每个文件中的每行返回一条记录。

• 对于 SequenceFiles，使用 SparkContext 的 sequenceFile[K, V] 方法，其中 K 和 V 是文件中键和值的类型。这些类型应该是 Hadoop Writable 接口的子类，例如 IntWritable 和 Text。

• 对于其他 Hadoop InputFormats，您可以使用 JavaSparkContext.hadoopRDD 方法，该方法接受任意的 JobConf、输入格式类、键类和值类。设置这些参数的方式与您在 Hadoop 作业中使用输入源的方式相同。您也可以使用 JavaSparkContext.newAPIHadoopRDD 处理基于“新”MapReduce API (org.apache.hadoop.mapreduce) 的 InputFormats。

• JavaRDD.saveAsObjectFile 和 JavaSparkContext.objectFile 支持以一种由序列化 Java 对象组成的简单格式保存 RDD。虽然这不如 Avro 等专用格式高效，但它提供了一种保存任何 RDD 的简单方法。

为了说明 RDD 的基础知识，请看下面的简单程序：

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

第一行定义了一个从外部文件读取的基础 RDD。此数据集既没有加载到内存中，也没有进行其他操作：lines 仅仅是一个指向文件的指针。第二行将 lineLengths 定义为 map 转换的结果。同样，由于惰性，lineLengths *没有*立即计算。最后，我们运行 reduce，这是一个行动。此时，Spark 将计算分解为在不同机器上运行的任务，每台机器都运行其 map 部分和局部归约，只将其结果返回给驱动程序。

如果我们稍后还想再次使用 lineLengths，我们可以添加：

lineLengths.persist()

在 reduce 之前，这将导致 lineLengths 在第一次计算后保存到内存中。

为了说明 RDD 的基础知识，请看下面的简单程序：

val lines = sc.textFile("data.txt")
val lineLengths = lines.map(s => s.length)
val totalLength = lineLengths.reduce((a, b) => a + b)

第一行定义了一个从外部文件读取的基础 RDD。此数据集既没有加载到内存中，也没有进行其他操作：lines 仅仅是一个指向文件的指针。第二行将 lineLengths 定义为 map 转换的结果。同样，由于惰性，lineLengths *没有*立即计算。最后，我们运行 reduce，这是一个行动。此时，Spark 将计算分解为在不同机器上运行的任务，每台机器都运行其 map 部分和局部归约，只将其结果返回给驱动程序。

如果我们稍后还想再次使用 lineLengths，我们可以添加：

lineLengths.persist()

在 reduce 之前，这将导致 lineLengths 在第一次计算后保存到内存中。

为了说明 RDD 的基础知识，请看下面的简单程序：

JavaRDD<String> lines = sc.textFile("data.txt");
JavaRDD<Integer> lineLengths = lines.map(s -> s.length());
int totalLength = lineLengths.reduce((a, b) -> a + b);

第一行定义了一个从外部文件读取的基础 RDD。此数据集既没有加载到内存中，也没有进行其他操作：lines 仅仅是一个指向文件的指针。第二行将 lineLengths 定义为 map 转换的结果。同样，由于惰性，lineLengths *没有*立即计算。最后，我们运行 reduce，这是一个行动。此时，Spark 将计算分解为在不同机器上运行的任务，每台机器都运行其 map 部分和局部归约，只将其结果返回给驱动程序。

如果我们稍后还想再次使用 lineLengths，我们可以添加：

lineLengths.persist(StorageLevel.MEMORY_ONLY());

在 reduce 之前，这将导致 lineLengths 在第一次计算后保存到内存中。

Spark 的 API 大量依赖于在驱动程序中传递函数以在集群上运行。有三种推荐的方法可以做到这一点：

• Lambda 表达式，用于可以写成表达式的简单函数。（Lambda 不支持多语句函数或不返回值的语句。）
• 在调用 Spark 的函数内部使用本地 def 定义，用于较长的代码。
• 模块中的顶级函数。

例如，要传递一个比 lambda 支持的更长的函数，请考虑以下代码：

"""MyScript.py"""
if __name__ == "__main__":
    def myFunc(s):
        words = s.split(" ")
        return len(words)

    sc = SparkContext(...)
    sc.textFile("file.txt").map(myFunc)

请注意，虽然也可以传递对类实例中方法（而不是单例对象）的引用，但这需要将包含该类的对象与方法一起发送。例如，考虑：

class MyClass(object):
    def func(self, s):
        return s
    def doStuff(self, rdd):
        return rdd.map(self.func)

在这里，如果我们创建一个 new MyClass 并对其调用 doStuff，那么内部的 map 引用的是 *该 MyClass 实例* 的 func 方法，因此整个对象都需要发送到集群。

类似地，访问外部对象的字段将引用整个对象：

class MyClass(object):
    def __init__(self):
        self.field = "Hello"
    def doStuff(self, rdd):
        return rdd.map(lambda s: self.field + s)

为了避免这个问题，最简单的方法是将 field 复制到局部变量中，而不是外部访问它：

def doStuff(self, rdd):
    field = self.field
    return rdd.map(lambda s: field + s)

Spark 的 API 大量依赖于在驱动程序中传递函数以在集群上运行。有两种推荐的方法可以做到这一点：

• 匿名函数语法，可用于短代码片段。
• 全局单例对象中的静态方法。例如，您可以定义 object MyFunctions，然后传递 MyFunctions.func1，如下所示：

object MyFunctions {
  def func1(s: String): String = { ... }
}

myRdd.map(MyFunctions.func1)

请注意，虽然也可以传递对类实例中方法（而不是单例对象）的引用，但这需要将包含该类的对象与方法一起发送。例如，考虑：

class MyClass {
  def func1(s: String): String = { ... }
  def doStuff(rdd: RDD[String]): RDD[String] = { rdd.map(func1) }
}

在这里，如果我们创建一个新的 MyClass 实例并对其调用 doStuff，那么内部的 map 引用的是 *该 MyClass 实例* 的 func1 方法，因此整个对象都需要发送到集群。这类似于编写 rdd.map(x => this.func1(x))。

类似地，访问外部对象的字段将引用整个对象：

class MyClass {
  val field = "Hello"
  def doStuff(rdd: RDD[String]): RDD[String] = { rdd.map(x => field + x) }
}

等效于编写 rdd.map(x => this.field + x)，这引用了整个 this。为了避免这个问题，最简单的方法是将 field 复制到局部变量中，而不是外部访问它：

def doStuff(rdd: RDD[String]): RDD[String] = {
  val field_ = this.field
  rdd.map(x => field_ + x)
}

Spark 的 API 大量依赖于在驱动程序中传递函数以在集群上运行。在 Java 中，函数由实现 org.apache.spark.api.java.function 包中接口的类表示。有两种创建此类函数的方法：

• 在您自己的类中实现函数接口，可以是匿名内部类或命名类，然后将其实例传递给 Spark。
• 使用 lambda 表达式来简洁地定义实现。

虽然本指南中的大部分内容都使用 lambda 语法以求简洁，但所有相同的 API 都可以轻松地以长形式使用。例如，我们可以将上面的代码写成如下形式：

JavaRDD<String> lines = sc.textFile("data.txt");
JavaRDD<Integer> lineLengths = lines.map(new Function<String, Integer>() {
  public Integer call(String s) { return s.length(); }
});
int totalLength = lineLengths.reduce(new Function2<Integer, Integer, Integer>() {
  public Integer call(Integer a, Integer b) { return a + b; }
});

或者，如果内联编写函数不方便：

class GetLength implements Function<String, Integer> {
  public Integer call(String s) { return s.length(); }
}
class Sum implements Function2<Integer, Integer, Integer> {
  public Integer call(Integer a, Integer b) { return a + b; }
}

JavaRDD<String> lines = sc.textFile("data.txt");
JavaRDD<Integer> lineLengths = lines.map(new GetLength());
int totalLength = lineLengths.reduce(new Sum());

请注意，Java 中的匿名内部类也可以访问封闭范围内的变量，只要它们被标记为 final。Spark 会将这些变量的副本发送到每个工作节点，就像它对其他语言所做的那样。

虽然大多数 Spark 操作适用于包含任何类型对象的 RDD，但少数特殊操作仅适用于键值对的 RDD。最常见的是分布式“混洗”操作，例如按键分组或聚合元素。

在 Python 中，这些操作适用于包含内置 Python 元组（如 (1, 2)）的 RDD。只需创建这样的元组，然后调用您想要的操作即可。

例如，以下代码使用 reduceByKey 操作在键值对上计算文件中每行文本出现的次数：

lines = sc.textFile("data.txt")
pairs = lines.map(lambda s: (s, 1))
counts = pairs.reduceByKey(lambda a, b: a + b)

我们还可以使用 counts.sortByKey()，例如，按字母顺序对对进行排序，最后使用 counts.collect() 将它们作为对象列表带回驱动程序。

虽然大多数 Spark 操作适用于包含任何类型对象的 RDD，但少数特殊操作仅适用于键值对的 RDD。最常见的是分布式“混洗”操作，例如按键分组或聚合元素。

在 Scala 中，这些操作自动适用于包含 Tuple2 对象的 RDD（语言中的内置元组，通过简单地编写 (a, b) 创建）。键值对操作在 PairRDDFunctions 类中可用，该类会自动包装元组的 RDD。

例如，以下代码使用 reduceByKey 操作在键值对上计算文件中每行文本出现的次数：

val lines = sc.textFile("data.txt")
val pairs = lines.map(s => (s, 1))
val counts = pairs.reduceByKey((a, b) => a + b)

我们还可以使用 counts.sortByKey()，例如，按字母顺序对对进行排序，最后使用 counts.collect() 将它们作为对象数组带回驱动程序。

注意：当在键值对操作中使用自定义对象作为键时，您必须确保自定义的 equals() 方法伴随有匹配的 hashCode() 方法。有关完整详细信息，请参阅 Object.hashCode() 文档中概述的契约。

虽然大多数 Spark 操作适用于包含任何类型对象的 RDD，但少数特殊操作仅适用于键值对的 RDD。最常见的是分布式“混洗”操作，例如按键分组或聚合元素。

在 Java 中，键值对使用 Scala 标准库中的 scala.Tuple2 类表示。您可以简单地调用 new Tuple2(a, b) 来创建一个元组，然后通过 tuple._1() 和 tuple._2() 访问其字段。

键值对的 RDD 由 JavaPairRDD 类表示。您可以使用 map 操作的特殊版本（如 mapToPair 和 flatMapToPair）从 JavaRDD 构建 JavaPairRDD。JavaPairRDD 将同时具有标准的 RDD 函数和特殊的键值函数。

例如，以下代码使用 reduceByKey 操作在键值对上计算文件中每行文本出现的次数：

JavaRDD<String> lines = sc.textFile("data.txt");
JavaPairRDD<String, Integer> pairs = lines.mapToPair(s -> new Tuple2(s, 1));
JavaPairRDD<String, Integer> counts = pairs.reduceByKey((a, b) -> a + b);

我们还可以使用 counts.sortByKey()，例如，按字母顺序对对进行排序，最后使用 counts.collect() 将它们作为对象数组带回驱动程序。

注意：当在键值对操作中使用自定义对象作为键时，您必须确保自定义的 equals() 方法伴随有匹配的 hashCode() 方法。有关完整详细信息，请参阅 Object.hashCode() 文档中概述的契约。

累加器是通过调用 SparkContext.accumulator(v) 从初始值 v 创建的。在集群上运行的任务可以使用 add 方法或 += 运算符向其添加值。但是，它们不能读取其值。只有驱动程序才能使用其 value 方法读取累加器的值。

以下代码展示了如何使用累加器对数组元素进行求和：

>>> accum = sc.accumulator(0)
>>> accum
Accumulator<id=0, value=0>

>>> sc.parallelize([1, 2, 3, 4]).foreach(lambda x: accum.add(x))
...
10/09/29 18:41:08 INFO SparkContext: Tasks finished in 0.317106 s

>>> accum.value
10

虽然此代码使用了对 Int 类型累加器的内置支持，但程序员也可以通过继承 AccumulatorParam 来创建自己的类型。AccumulatorParam 接口有两个方法：zero 用于为您的数据类型提供一个“零值”，以及 addInPlace 用于将两个值相加。例如，假设我们有一个表示数学向量的 Vector 类，我们可以这样编写

class VectorAccumulatorParam(AccumulatorParam):
    def zero(self, initialValue):
        return Vector.zeros(initialValue.size)

    def addInPlace(self, v1, v2):
        v1 += v2
        return v1

# Then, create an Accumulator of this type:
vecAccum = sc.accumulator(Vector(...), VectorAccumulatorParam())

可以通过调用 SparkContext.longAccumulator() 或 SparkContext.doubleAccumulator() 来创建数值累加器，分别累加 Long 或 Double 类型的值。在集群上运行的任务可以使用 add 方法向其添加值。然而，它们无法读取其值。只有驱动程序可以使用其 value 方法读取累加器的值。

以下代码展示了如何使用累加器对数组元素进行求和：

scala> val accum = sc.longAccumulator("My Accumulator")
accum: org.apache.spark.util.LongAccumulator = LongAccumulator(id: 0, name: Some(My Accumulator), value: 0)

scala> sc.parallelize(Array(1, 2, 3, 4)).foreach(x => accum.add(x))
...
10/09/29 18:41:08 INFO SparkContext: Tasks finished in 0.317106 s

scala> accum.value
res2: Long = 10

虽然此代码使用了对 Long 类型累加器的内置支持，但程序员也可以通过继承 AccumulatorV2 来创建自己的类型。AccumulatorV2 抽象类有几个必须重写的方法：reset 用于将累加器重置为零，add 用于将另一个值添加到累加器中，merge 用于将另一个同类型累加器合并到此累加器中。其他必须重写的方法包含在 API 文档中。例如，假设我们有一个表示数学向量的 MyVector 类，我们可以这样编写

class VectorAccumulatorV2 extends AccumulatorV2[MyVector, MyVector] {

  private val myVector: MyVector = MyVector.createZeroVector

  def reset(): Unit = {
    myVector.reset()
  }

  def add(v: MyVector): Unit = {
    myVector.add(v)
  }
  ...
}

// Then, create an Accumulator of this type:
val myVectorAcc = new VectorAccumulatorV2
// Then, register it into spark context:
sc.register(myVectorAcc, "MyVectorAcc1")

请注意，当程序员定义自己的 AccumulatorV2 类型时，结果类型可以与添加的元素类型不同。

可以通过调用 SparkContext.longAccumulator() 或 SparkContext.doubleAccumulator() 来创建数值累加器，分别累加 Long 或 Double 类型的值。在集群上运行的任务可以使用 add 方法向其添加值。然而，它们无法读取其值。只有驱动程序可以使用其 value 方法读取累加器的值。

以下代码展示了如何使用累加器对数组元素进行求和：

LongAccumulator accum = jsc.sc().longAccumulator();

sc.parallelize(Arrays.asList(1, 2, 3, 4)).foreach(x -> accum.add(x));
// ...
// 10/09/29 18:41:08 INFO SparkContext: Tasks finished in 0.317106 s

accum.value();
// returns 10

虽然此代码使用了对 Long 类型累加器的内置支持，但程序员也可以通过继承 AccumulatorV2 来创建自己的类型。AccumulatorV2 抽象类有几个必须重写的方法：reset 用于将累加器重置为零，add 用于将另一个值添加到累加器中，merge 用于将另一个同类型累加器合并到此累加器中。其他必须重写的方法包含在 API 文档中。例如，假设我们有一个表示数学向量的 MyVector 类，我们可以这样编写

class VectorAccumulatorV2 implements AccumulatorV2<MyVector, MyVector> {

  private MyVector myVector = MyVector.createZeroVector();

  public void reset() {
    myVector.reset();
  }

  public void add(MyVector v) {
    myVector.add(v);
  }
  ...
}

// Then, create an Accumulator of this type:
VectorAccumulatorV2 myVectorAcc = new VectorAccumulatorV2();
// Then, register it into spark context:
jsc.sc().register(myVectorAcc, "MyVectorAcc1");

请注意，当程序员定义自己的 AccumulatorV2 类型时，结果类型可以与添加的元素类型不同。

警告：当 Spark 任务完成时，Spark 将尝试将此任务中累积的更新合并到一个累加器中。如果失败，Spark 将忽略失败并仍然将任务标记为成功，并继续运行其他任务。因此，一个有缺陷的累加器不会影响 Spark 作业，但即使 Spark 作业成功，它也可能无法正确更新。