Spark 3.5.1 与 Python 3.8+ 兼容。它可以使用标准的 CPython 解释器，因此可以使用 NumPy 等 C 库。它也与 PyPy 7.3.6+ 兼容。

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

    install_requires=[
        'pyspark=={site.SPARK_VERSION}'
    ]

要运行没有通过 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 3.5.1 默认构建并分发为与 Scala 2.12 兼容。（Spark 也可以构建为与其他版本的 Scala 兼容。）要使用 Scala 编写应用程序，你需要使用兼容的 Scala 版本（例如 2.12.X）。

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

groupId = org.apache.spark
artifactId = spark-core_2.12
version = 3.5.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 3.5.1 支持 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.12
version = 3.5.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 参数设置上下文连接到的主节点，并且可以通过将逗号分隔的列表传递给 --py-files 参数来将 Python .zip、.egg 或 .py 文件添加到运行时路径。对于第三方 Python 依赖项，请参阅 Python 包管理。你也可以通过将逗号分隔的 Maven 坐标列表传递给 --packages 参数来将依赖项（例如 Spark 包）添加到你的 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 参数设置上下文连接到的主节点，并且可以通过将逗号分隔的列表传递给 --jars 参数来将 JAR 添加到类路径中。你也可以通过将逗号分隔的 Maven 坐标列表传递给 --packages 参数来将依赖项（例如 Spark 包）添加到你的 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 形成对比，textFile 将为每个文件中的每一行返回一个记录。

• RDD.saveAsPickleFile 和 SparkContext.pickleFile 支持将 RDD 保存为由腌制 Python 对象组成的简单格式。在腌制序列化上使用批处理，默认批处理大小为 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[]，然后将其腌制为 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 特性。只需扩展此特性并在 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 作业中使用输入源的方式相同。您还可以使用 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 允许您读取包含多个小型文本文件的目录，并将其中的每个文件作为 (文件名，内容) 对返回。这与 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 包中的接口的类表示。有两种方法可以创建此类函数

• 在您自己的类中实现 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() 文档 中概述的契约。

累加器是从初始值 v 创建的，方法是调用 SparkContext.accumulator(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 作业成功了。