Spark 3.5.1 可搭配 Python 3.8+ 使用。它可以使用標準的 CPython 解譯器，因此可以搭配 NumPy 等 C 函式庫使用。它也可以搭配 PyPy 7.3.6+ 使用。

Python 中的 Spark 應用程式可以搭配 bin/spark-submit 執行，它包含執行時期的 Spark，或將其包含在您的 setup.py 中，如下所示

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

若要在不安裝 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 參數設定 context 連線的 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 中啟動 PySpark shell，這是增強的 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 伺服器後，您可以從「檔案」標籤建立新的筆記本。在筆記本中，您可以在開始從 Jupyter notebook 嘗試 Spark 之前，將命令 %pylab inline 作為筆記本的一部分輸入。

在 Spark shell 中，一個特別的解釋器感知 SparkContext 已為您建立，在變數 sc 中。建立您自己的 SparkContext 將無法運作。您可以使用 --master 參數設定 context 連線的 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 方法，對現有的 iterable 或集合建立。集合的元素會複製以形成一個分布式資料集，可以並行運算。例如，以下是如何建立一個包含數字 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 支援以包含醃製 Python 物件的簡單格式儲存 RDD。批次處理用於醃製序列化，預設批次大小為 10。

• SequenceFile 和 Hadoop 輸入/輸出格式

注意此功能目前標記為 實驗性，並供進階使用者使用。它可能會在未來被基於 Spark SQL 的讀取/寫入支援取代，在這種情況下，Spark SQL 是首選方法。

可寫入支援

PySpark SequenceFile 支援在 Java 中載入 RDD 的鍵值對，將 Writable 轉換為基本 Java 類型，並使用 pickle 將產生的 Java 物件序列化。當將 RDD 的鍵值對儲存到 SequenceFile 時，PySpark 會執行相反的動作。它會將 Python 物件反序列化為 Java 物件，然後將它們轉換為 Writable。下列 Writable 會自動轉換

陣列並未預設處理。使用者需要在讀取或寫入時指定自訂 ArrayWritable 子類型。在寫入時，使用者也需要指定自訂轉換器，將陣列轉換為自訂 ArrayWritable 子類型。在讀取時，預設轉換器會將自訂 ArrayWritable 子類型轉換為 Java Object[]，然後將其序列化為 Python 元組。若要取得基本類型陣列的 Python array.array，使用者需要指定自訂轉換器。

儲存和載入 SequenceFile

類似於文字檔案，SequenceFile 可以透過指定路徑來儲存和載入。可以指定鍵和值類別，但對於標準 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 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 特性可供使用。只要延伸此特性，並在 convert 方法中實作轉換程式碼即可。請務必確保此類別連同存取 InputFormat 所需的任何依賴項，都打包到您的 Spark 任務 jar 中，並包含在 PySpark classpath 中。

請參閱 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 允許您為一些常見的 Writable 指定原生類型；例如，sequenceFile[Int, String] 會自動讀取 IntWritable 和 Text。

• 對於其他 Hadoop InputFormat，您可以使用 SparkContext.hadoopRDD 方法，它會採用任意的 JobConf 和輸入格式類別、金鑰類別和值類別。請以與使用輸入來源的 Hadoop 工作相同的設定方式設定這些。您也可以使用 SparkContext.newAPIHadoopRDD，針對基於「新」MapReduce API (org.apache.hadoop.mapreduce) 的 InputFormat。

• 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 會將計算分解成在個別機器上執行的任務，而每部機器都會執行其對應的地圖部分和區域性縮減，只傳回其答案給驅動程式。

如果我們稍後也想再次使用 lineLengths，我們可以在 reduce 之前新增

lineLengths.persist()

，這會導致 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，我們可以在 reduce 之前新增

lineLengths.persist()

，這會導致 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，我們可以在 reduce 之前新增

lineLengths.persist(StorageLevel.MEMORY_ONLY());

，這會導致 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。最常見的是分散式「洗牌」作業，例如依據某個鍵對元素進行分組或聚合。

在 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 物件 (語言中的內建元組，透過簡單撰寫 (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。最常見的是分散式「洗牌」作業，例如依據某個鍵對元素進行分組或聚合。

在 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 工作成功，它也可能無法正確更新。