Spark 3.5.3 与 Python 3.8+ 兼容。它可以使用标准的 CPython 解释器，因此可以使用像 NumPy 这样的 C 库。它也支持 PyPy 7.3.6+。

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

 install_requires=[
'pyspark==3.5.3'
]

要在Python中运行Spark应用程序而无需使用pip安装PySpark，请使用位于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.3 默认构建并分发以与 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.3

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

groupId = org.apache.hadoop
artifactId = hadoop-client
version = 

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

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

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

Spark 3.5.3 支持 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.3

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

groupId = org.apache.hadoop
artifactId = hadoop-client
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 参数设置上下文连接的主节点，并通过传递以逗号分隔的列表将Python .zip、.egg或.py文件添加到运行时路径中，使用 --py-files 。有关第三方Python依赖项，请参见 Python包管理 。您还可以通过向 --packages 参数提供以逗号分隔的Maven坐标列表，将依赖项（例如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 中启动 PySpark shell，IPython 是增强版的 Python 解释器。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服务器启动后，您可以从“文件”标签创建一个新笔记本。在笔记本中，您可以输入命令 %pylab inline ，作为您在开始尝试从Jupyter笔记本使用Spark之前的一部分。

在Spark Shell中，已经为您创建了一个特殊的解释器感知的SparkContext，存储在名为 sc 的变量中。自己创建SparkContext将不起作用。您可以使用 --master 参数设置上下文连接的主节点，并通过传递一个以逗号分隔的列表将JAR添加到类路径中，使用 --jars 参数。您还可以通过提供一个以逗号分隔的Maven坐标列表，将依赖项（例如Spark包）添加到您的Shell会话中，使用 --packages 参数。任何可能存在依赖项的其他存储库（例如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) 来将列表的元素相加。 我们将在后面描述对分布式数据集的操作。

并行化集合是通过调用 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 。

文本文件 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 支持以简单格式保存 RDD，该格式由序列化的 Python 对象组成。序列化时使用批处理，默认批量大小为 10。

• SequenceFile 和 Hadoop 输入/输出格式

注意 该功能目前标记为 实验性 ，旨在供高级用户使用。未来可能会被基于 Spark SQL 的读/写支持替代，在这种情况下，Spark SQL 将是首选方法。

可写支持

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

数组不能直接处理。用户在读取或写入时需要指定自定义 ArrayWritable 子类型。在写入时，用户还需要指定将数组转换成自定义 ArrayWritable 子类型的自定义转换器。在读取时，默认转换器将自定义 ArrayWritable 子类型转换为 Java Object[] ，然后被序列化为 Python 元组。为了获得原始类型数组的 Python array.array ，用户需要指定自定义转换器。

保存和加载序列文件

与文本文件类似，可以通过指定路径保存和加载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"} # 假设Elasticsearch在localhost默认运行
>>> rdd = sc.newAPIHadoopRDD("org.elasticsearch.hadoop.mr.EsInputFormat",
"org.apache.hadoop.io.NullWritable",
"org.elasticsearch.hadoop.mr.LinkedMapWritable",
conf=conf)
>>> rdd.first() # 结果是一个MapWritable，转换为Python字典
(u'Elasticsearch ID',
{u: True,
u: u,
u: 12345})

请注意，如果 InputFormat 仅依赖于 Hadoop 配置和/或输入路径，并且键和值类可以根据上述表格轻松转换，那么这种方法在此类情况下应该效果良好。

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

查看 Python 示例 和 转换器示例 ，了解如何使用 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] 在 textFile 在 <控制台>: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 支持以简单格式保存 RDD，该格式由序列化的 Java 对象组成。虽然这不如专门的格式（如 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 支持以简单格式保存 RDD，该格式由序列化的 Java 对象组成。虽然这不如像 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将计算拆分为任务 在不同的机器上运行，且每台机器同时运行其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 引用的是 func 方法 该 MyClass 实例的 ，因此整个对象需要被发送到集群。

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

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 引用了 func1 方法 该 MyClass 实例的 ，因此整个对象需要被发送到集群。这类似于写 rdd.map(x => this.func1(x)) 。

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

class MyClass {
val field = "你好"
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元组的RDD，例如 (1, 2) 。只需创建这样的元组，然后调用所需的操作。

例如，以下代码使用 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.Tuple2 类 来自Scala标准库。您可以简单地调用 new Tuple2(a, b) 来创建一个元组，并稍后使用 tuple._1() 和 tuple._2() 访问其字段。

键值对的RDD由 JavaPairRDD 类表示。你可以使用特殊版本的 map 操作，从JavaRDD构建JavaPairRDD，比如 mapToPair 和 flatMapToPair 。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
累加器<id=0, value=0>
>>> sc.parallelize([1, 2, 3, 4]).foreach(lambda x: accum.add(x))
...
10/09/29 18:41:08 信息 SparkContext: 任务 完成 在 0.317106 秒
>>> 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
# 然后，创建一个这种类型的累加器：
vecAccum = sc.accumulator(Vector(...), VectorAccumulatorParam())

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

下面的代码显示了一个累加器用来将数组的元素相加：

scala> val accum = sc.longAccumulator("我的累加器")
accum: org.apache.spark.util.LongAccumulator = LongAccumulator(id: 0, name: Some(我的 累加器), value: 0)
scala> sc.parallelize(Array(1, 2, 3, 4)).foreach(x => accum.add(x))
...
10/09/29 18:41:08 信息 SparkContext: 任务 完成 在 0.317106 秒
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)
}
...
}
// 然后，创建这种类型的累加器：
val myVectorAcc = new VectorAccumulatorV2
// 然后，将其注册到spark上下文中：
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 信息 SparkContext: 任务在 0.317106 秒内完成
accum.value();
// 返回 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);
}
...
}
// 然后，创建这种类型的累加器：
VectorAccumulatorV2 myVectorAcc = new VectorAccumulatorV2();
// 然后，将其注册到 spark 上下文中：
jsc.sc().register(myVectorAcc, "MyVectorAcc1");

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

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