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

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

    install_requires=[
        'pyspark==3.5.5'
    ]

要在没有 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 要求 driver 和 worker 使用相同的 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.5 默认构建和分发以与 Scala 2.12 配合使用。（Spark 也可以构建为与其他版本的 Scala 配合使用。）要使用 Scala 编写应用程序，您需要使用兼容的 Scala 版本（例如 2.12.X）。

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

groupId = org.apache.spark
artifactId = spark-core_2.12
version = 3.5.5

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

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

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

groupId = org.apache.spark
artifactId = spark-core_2.12
version = 3.5.5

此外，如果您希望访问 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 包管理。 您还可以通过将 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 服务器后，您可以从“Files”（文件）选项卡创建一个新的 notebook。在 notebook 内部，您可以在开始尝试从 Jupyter notebook 使用 Spark 之前，输入命令 %pylab inline 作为 notebook 的一部分。

在 Spark shell 中，已经为您创建了一个特殊的、支持解释器的 SparkContext，它存储在名为 sc 的变量中。您自己创建 SparkContext 将无法工作。您可以使用 --master 参数设置上下文连接到的 master，并且可以通过将逗号分隔的列表传递给 --jars 参数来将 JAR 添加到 classpath 中。您还可以通过将逗号分隔的 Maven 坐标列表提供给 --packages 参数，将依赖项（例如，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) 来将列表中的元素相加。我们将在后面描述分布式数据集上的操作。

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

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

创建后，可以并行操作分布式数据集（distData）。例如，我们可以调用 distData.reduce((a, b) => a + b) 来将数组中的元素相加。我们将在后面描述分布式数据集上的操作。

并行化集合是通过在驱动程序中使用 JavaSparkContext 的 parallelize 方法对现有的 Collection 进行调用来创建的。集合的元素被复制以形成可以并行操作的分布式数据集。例如，以下是如何创建一个包含数字 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。

可以使用 SparkContext 的 textFile 方法创建文本文件 RDD。此方法接受文件的 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 读取文件的一些注意事项：

• 如果使用本地文件系统上的路径，则该文件还必须在 worker 节点上的相同路径上可访问。可以将文件复制到所有 worker，或者使用网络挂载的共享文件系统。

• 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 不同，后者将为每个文件中的每一行返回一条记录。

• RDD.saveAsPickleFile 和 SparkContext.pickleFile 支持以一种简单的格式保存 RDD，该格式由 pickled Python 对象组成。批量处理用于 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 dict 传入。这是一个使用 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。

可以使用 SparkContext 的 textFile 方法创建文本文件 RDD。此方法接受文件的 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 读取文件的一些注意事项：

• 如果使用本地文件系统上的路径，则该文件还必须在 worker 节点上的相同路径上可访问。可以将文件复制到所有 worker，或者使用网络挂载的共享文件系统。

• 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 不同，后者将为每个文件中的每一行返回一条记录。 分区由数据本地性决定，在某些情况下，可能导致分区过少。 对于这些情况，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 用于基于“new” 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。

可以使用 SparkContext 的 textFile 方法创建文本文件 RDD。此方法接受文件的 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 读取文件的一些注意事项：

• 如果使用本地文件系统上的路径，则该文件还必须在 worker 节点上的相同路径上可访问。可以将文件复制到所有 worker，或者使用网络挂载的共享文件系统。

• 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 不同，后者将为每个文件中的每一行返回一条记录。

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

• 对于其他 Hadoop InputFormats，您可以使用 JavaSparkContext.hadoopRDD 方法，该方法接受任意 JobConf 和输入格式类、键类和值类。 按照与使用输入源的 Hadoop 作业相同的方式设置这些。 您还可以使用 JavaSparkContext.newAPIHadoopRDD 用于基于“new” 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 和本地 reduce，只将其答案返回给驱动程序。

如果我们以后还想再次使用 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 和本地 reduce，只将其答案返回给驱动程序。

如果我们以后还想再次使用 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 和本地 reduce，只将其答案返回给驱动程序。

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

lineLengths.persist(StorageLevel.MEMORY_ONLY());

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

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

• Lambda 表达式，用于可以写成表达式的简单函数。 （Lambdas 不支持多语句函数或不返回值语句。）
• 调用 Spark 的函数内部的局部 defs，用于更长的代码。
• 模块中的顶级函数。

例如，要传递比使用 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 会将这些变量的副本发送到每个 worker 节点，就像对其他语言一样。

虽然大多数 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 方法或 += 运算符向其添加内容。但是，它们无法读取其值。只有驱动程序可以使用其 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 作业成功，它也可能无法正确更新。