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

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

    install_requires=[
        'pyspark==4.1.1'
    ]

要在不使用 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.1.1 默认构建并分发以支持 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.1.1

此外，如果您希望访问 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.1.1 支持 Lambda 表达式 以简洁地编写函数，否则您可以使用 org.apache.spark.api.java.function 包中的类。

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

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

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

此外，如果您希望访问 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 服务器启动后，您可以从“Files”选项卡创建一个新的 notebook。在 notebook 中，您可以输入命令 %pylab inline 作为 notebook 的一部分，然后再开始尝试从 Jupyter notebook 中使用 Spark。

在 Spark shell 中，已经为您创建了一个特殊的、可识别解释器的 SparkContext，存储在变量 sc 中。自行创建 SparkContext 将无法工作。您可以使用 --master 参数设置上下文连接的 master，并且可以通过传递逗号分隔的列表给 --jars 参数将 JAR 文件添加到类路径中。您还可以通过向 --packages 参数提供逗号分隔的 Maven 坐标列表，将依赖项（例如 Spark Packages）添加到您的 shell 会话中。可以存在依赖项的任何额外存储库（例如 Sonatype）都可以传递给 --repositories 参数。例如，要在一个四核上运行 bin/spark-shell，请使用

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

或者，要同时将 code.jar 添加到其类路径中，请使用

$ ./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 允许您读取包含多个小文本文件的目录，并将它们中的每一个作为 (文件名, 内容) 对返回。这与 textFile 形成对比，后者会为每个文件中的每一行返回一条记录。

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

• SequenceFile 和 Hadoop 输入/输出格式

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

Writable 支持

PySpark SequenceFile 支持加载 Java 中的键值对 RDD，将 Writables 转换为基础 Java 类型，并使用 pickle 对生成的 Java 对象进行 pickle 处理。当将键值对 RDD 保存到 SequenceFile 时，PySpark 执行相反的操作。它将 Python 对象 unpickle 为 Java 对象，然后将它们转换为 Writables。以下 Writables 会自动转换

数组无法直接开箱即用。用户在读取或写入时需要指定自定义 ArrayWritable 子类型。写入时，用户还需要指定将数组转换为自定义 ArrayWritable 子类型的自定义转换器。读取时，默认转换器会将自定义 ArrayWritable 子类型转换为 Java Object[]，然后将其 pickle 到 Python 元组。要获取基本类型数组的 Python array.array，用户需要指定自定义转换器。

保存和加载 SequenceFiles

与文本文件类似，可以通过指定路径来保存和加载 SequenceFiles。可以指定键和值类，但对于标准 Writables，这不是必需的。

>>> 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)。只需扩展此特质并在 convert 方法中实现您的转换代码。请记住确保此类以及访问您的 InputFormat 所需的任何依赖项都打包到您的 Spark 作业 jar 中，并包含在 PySpark 类路径中。

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

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 允许您读取包含多个小文本文件的目录，并将它们中的每一个作为 (文件名, 内容) 对返回。这与 textFile 形成对比，后者会为每个文件中的每一行返回一条记录。分区由数据本地性决定，在某些情况下，这可能导致分区过少。对于这些情况，wholeTextFiles 提供了一个可选的第二个参数来控制分区的最小数量。

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

• 对于其他 Hadoop InputFormats，您可以使用 SparkContext.hadoopRDD 方法，它接受任意 JobConf 和输入格式类、键类和值类。按照您为带有输入源的 Hadoop 作业设置它们的方式来设置这些参数。您还可以对基于“新” MapReduce API (org.apache.hadoop.mapreduce) 的 InputFormats 使用 SparkContext.newAPIHadoopRDD。

• 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 允许您读取包含多个小文本文件的目录，并将它们中的每一个作为 (文件名, 内容) 对返回。这与 textFile 形成对比，后者会为每个文件中的每一行返回一条记录。

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

• 对于其他 Hadoop InputFormats，您可以使用 JavaSparkContext.hadoopRDD 方法，它接受任意 JobConf 和输入格式类、键类和值类。按照您为带有输入源的 Hadoop 作业设置它们的方式来设置这些参数。您还可以对基于“新” MapReduce API (org.apache.hadoop.mapreduce) 的 InputFormats 使用 JavaSparkContext.newAPIHadoopRDD。

• 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 将计算拆分为在不同机器上运行的任务，每台机器运行其映射部分和本地缩减，并将答案仅返回给驱动程序。

如果我们以后也想再次使用 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 将计算拆分为在不同机器上运行的任务，每台机器运行其映射部分和本地缩减，并将答案仅返回给驱动程序。

如果我们以后也想再次使用 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 将计算拆分为在不同机器上运行的任务，每台机器运行其映射部分和本地缩减，并将答案仅返回给驱动程序。

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

lineLengths.persist(StorageLevel.MEMORY_ONLY());

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

Spark 的 API 严重依赖于在驱动程序中传递函数以在集群上运行。有三种推荐的方法来执行此操作

• Lambda 表达式，适用于可以编写为表达式的简单函数。（Lambda 不支持多语句函数或不返回值的语句。）
• 函数内部的本地 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 包中接口的类来表示。有两种方法可以创建此类函数

• 在您自己的类中实现 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。最常见的是分布式“shuffle”操作，例如按键分组或聚合元素。

在 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。最常见的是分布式“shuffle”操作，例如按键分组或聚合元素。

在 Scala 中，这些操作自动适用于包含 Tuple2 对象（语言中的内置元组，只需编写 (a, b) 即可创建）的 RDD。键值对操作可在 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。最常见的是分布式“shuffle”操作，例如按键分组或聚合元素。

在 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 方法或 += 操作符对其进行累加。但是，任务无法读取其值。只有驱动程序（driver program）可以使用其 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 方法对其进行累加。但是，任务无法读取其值。只有驱动程序（driver program）可以使用其 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 方法对其进行累加。但是，任务无法读取其值。只有驱动程序（driver program）可以使用其 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 会忽略该失败，仍然将任务标记为成功，并继续运行其他任务。因此，有 Bug 的累加器不会影响 Spark 作业，但尽管 Spark 作业成功，累加器可能未被正确更新。