Spark 安裝部署與快速上手

• 2021 年 11 月 18 日
• 筆記
• spark, Big Data

Spark 介紹

核心概念

Spark 是 UC Berkeley AMP lab 開發的一個集群計算的框架，類似於 Hadoop，但有很多的區別。

最大的優化是讓計算任務的中間結果可以存儲在記憶體中，不需要每次都寫入 HDFS，更適用於需要迭代的 MapReduce 演算法場景中，可以獲得更好的性能提升。

例如一次排序測試中，對 100TB 數據進行排序，Spark 比 Hadoop 快三倍，並且只需要十分之一的機器。

Spark 集群目前最大的可以達到 8000 節點，處理的數據達到 PB 級別，在互聯網企業中應用非常廣泛。

Spark 的特性

Hadoop 的核心是分散式文件系統 HDFS 和計算框架 MapReduces。Spark 可以替代 MapReduce，並且兼容 HDFS、Hive 等分散式存儲層，良好的融入 Hadoop 的生態系統。

Spark 執行的特點

• 中間結果輸出：Spark 將執行工作流抽象為通用的有向無環圖執行計劃（DAG），可以將多 Stage 的任務串聯或者並行執行。
• 數據格式和記憶體布局：Spark 抽象出分散式記憶體存儲結構彈性分散式數據集 RDD，能夠控制數據在不同節點的分區，用戶可以自定義分區策略。
• 任務調度的開銷：Spark 採用了事件驅動的類庫 AKKA 來啟動任務，通過執行緒池的復用執行緒來避免系統啟動和切換開銷。

Spark 的優勢

• 速度快，運行工作負載快 100 倍。Apache Spark 使用最先進的 DAG 調度器、查詢優化器和物理執行引擎，實現了批處理和流數據的高性能。
• 易於使用，支援用 Java、Scala、Python、R 和 SQL 快速編寫應用程式。Spark 提供了超過 80 個運算元，可以輕鬆構建並行應用程式。您可以從 Scala、Python、R 和 SQL shell 中互動式地使用它。
• 普遍性，結合 SQL、流處理和複雜分析。Spark 提供了大量的庫，包括 SQL 和 DataFrames、用於機器學習的 MLlib、GraphX 和 Spark 流。您可以在同一個應用程式中無縫地組合這些庫。
• 各種環境都可以運行，Spark 在 Hadoop、Apache Mesos、Kubernetes、單機或雲主機中運行。它可以訪問不同的數據源。您可以使用它的獨立集群模式在 EC2、Hadoop YARN、Mesos 或 Kubernetes 上運行 Spark。訪問 HDFS、Apache Cassandra、Apache HBase、Apache Hive 和數百個其他數據源中的數據。

哪些公司在使用 Spark

日常為我們所熟知的，在國外就有 IBM Almaden（IBM 研究實驗室）、Amazon（亞馬遜）等，而在中國有 baidu（百度）、Tencent（騰訊）等等，包括一些其它的公司大部分都使用 Spark 來處理生產過程中產生的大量數據。更多詳情可以參考鏈接： 誰在使用 Spark?

2.3 Spark 生態系統 BDAS

目前，Spark 已經發展成為包含眾多子項目的大數據計算平台。

BDAS 是伯克利大學提出的基於 Spark 的數據分析棧（BDAS）。

其核心框架是 Spark，同時涵蓋支援結構化數據 SQL 查詢與分析的查詢引擎 Spark SQL，提供機器學習功能的系統 MLBase 及底層的分散式機器學習庫 MLlib，並行圖計算框架 GraphX，流計算框架 Spark Streaming，近似查詢引擎 BlinkDB，記憶體分散式文件系統 Tachyon，資源管理框架 Mesos 等子項目。這些子項目在 Spark 上層提供了更高層、更豐富的計算範式。

部署前準備

Spark 安裝非常簡單，簡單到只需要下載 binary 包解壓即可

安裝 Spark 之前需要先安裝 Java，Scala 及 Python。

• java 1.8.0
• scala 2.11.8
• python 2.7

安裝Java

下載

參考清華軟體源：//mirrors.tuna.tsinghua.edu.cn/AdoptOpenJDK/

wget wget //mirrors.tuna.tsinghua.edu.cn/AdoptOpenJDK/8/jdk/x64/linux/OpenJDK8U-jdk_x64_linux_openj9_linuxXL_8u282b08_openj9-0.24.0.tar.gz --no-check-certificate

按需求下載之後記得修改環境變數

記得配置環境變數，添加 bin 目錄即可

安裝Scala

下載

wget //downloads.lightbend.com/scala/2.11.8/scala-2.11.8.tgz

記得配置環境變數，添加 bin 目錄即可

安裝Python

一般系統自帶 Python2

python --version

如果需要 Python3 可以自行下載

yum -y install python3

Spark 下載

此處使用的是：Spark 2.4.8

官網上下載已經預編譯好的 Spark binary，直接解壓即可。

Spark 官方下載鏈接：//spark.apache.org/downloads.html

下載

參考

• 清華軟體源：//mirrors.tuna.tsinghua.edu.cn/apache/spark/
• 華為軟體源：//mirrors.huaweicloud.com/apache/spark/

wget //mirrors.huaweicloud.com/apache/spark/spark-2.4.4/spark-2.4.4-bin-hadoop2.7.tgz

記得配置環境變數，添加 bin 目錄即可

部署模式介紹

Spark on Mesos未嘗試過，大家可以自行嘗試

本文僅介紹 Standalone 模式和 Spark on Yarn模式

按照自己需求配合！

本地模式

Spark單機運行，直接解壓執行start-all.sh即可，一般用於開發測試

Standalone 模式

構建一個由Master+Slave構成的Spark集群，Spark運行在集群中。

即獨立模式，自帶完整的服務，可單獨部署到一個集群中，無需依賴任何其他資源管理系統。

從一定程度上說，該模式是其他兩種的基礎。借鑒 Spark 開發模式，我們可以得到一種開發新型計算框架的一般思路：先設計出它的 standalone 模式，為了快速開發，起初不需要考慮服務（比如 master/slave）的容錯性，之後再開發相應的 wrapper，將 stanlone 模式下的服務原封不動的部署到資源管理系統 yarn 或者 mesos 上，由資源管理系統負責服務本身的容錯。目前 Spark 在 standalone 模式下是沒有任何單點故障問題的，這是藉助 zookeeper 實現的，思想類似於 Hbase master 單點故障解決方案。將 Spark standalone 與 MapReduce 比較，會發現它們兩個在架構上是完全一致的：

1. 都是由 master/slaves 服務組成的，且起初 master 均存在單點故障，後來均通過 zookeeper 解決（Apache MRv1 的 JobTracker 仍存在單點問題，但 CDH 版本得到了解決）；
2. 各個節點上的資源被抽象成粗粒度的 slot，有多少 slot 就能同時運行多少 task。不同的是，MapReduce 將 slot 分為 map slot 和 reduce slot，它們分別只能供 Map Task 和 Reduce Task 使用，而不能共享，這是 MapReduce 資源利率低效的原因之一，而 Spark 則更優化一些，它不區分 slot 類型，只有一種 slot，可以供各種類型的 Task 使用，這種方式可以提高資源利用率，但是不夠靈活，不能為不同類型的 Task 訂製 slot 資源。總之，這兩種方式各有優缺點。

Spark on Yarn 模式

Spark客戶端直接連接Yarn。不需要額外構建Spark集群。

這是一種很有前景的部署模式。但限於 YARN 自身的發展，目前僅支援粗粒度模式（Coarse-grained Mode）。

這是由於 YARN 上的 Container 資源是不可以動態伸縮的，一旦 Container 啟動之後，可使用的資源不能再發生變化，不過這個已經在 YARN 計劃中了。

spark on yarn 的支援兩種模式：

1. yarn-cluster：適用於生產環境；
2. yarn-client：適用於交互、調試，希望立即看到 app 的輸出

yarn-cluster 和 yarn-client 的區別在於 yarn appMaster，每個 yarn app 實例有一個 appMaster 進程，是為 app 啟動的第一個 container；負責從 ResourceManager 請求資源，獲取到資源後，告訴 NodeManager 為其啟動 container。

yarn-cluster 和 yarn-client 模式內部實現還是有很大的區別。

如果你需要用於生產環境，那麼請選擇 yarn-cluster；而如果你僅僅是 Debug 程式，可以選擇 yarn-client。

Spark on Mesos 模式

Spark客戶端直接連接Mesos。不需要額外構建Spark集群。

這是很多公司採用的模式，官方推薦這種模式（當然，原因之一是血緣關係）。

正是由於 Spark 開發之初就考慮到支援 Mesos，因此，目前而言，Spark 運行在 Mesos 上會比運行在 YARN 上更加靈活，更加自然。

目前在 Spark On Mesos 環境中，用戶可選擇兩種調度模式之一運行自己的應用程式（可參考 Andrew Xia 的「Mesos Scheduling Mode on Spark」）：

1. 粗粒度模式（Coarse-grained Mode）：每個應用程式的運行環境由一個 Dirver 和若干個 Executor 組成，其中，每個 Executor 佔用若干資源，內部可運行多個 Task（對應多少個「slot」）。

   應用程式的各個任務正式運行之前，需要將運行環境中的資源全部申請好，且運行過程中要一直佔用這些資源，即使不用，最後程式運行結束後，回收這些資源。舉個例子，比如你提交應用程式時，指定使用 5 個 executor 運行你的應用程式，每個 executor 佔用 5GB 記憶體和 5 個 CPU，每個 executor 內部設置了 5 個 slot，則 Mesos 需要先為 executor 分配資源並啟動它們，之後開始調度任務。

   另外，在程式運行過程中，mesos 的 master 和 slave 並不知道 executor 內部各個 task 的運行情況，executor 直接將任務狀態通過內部的通訊機制彙報給 Driver，從一定程度上可以認為，每個應用程式利用 mesos 搭建了一個虛擬集群自己使用。

2. 細粒度模式（Fine-grained Mode）：鑒於粗粒度模式會造成大量資源浪費，Spark On Mesos 還提供了另外一種調度模式：細粒度模式，這種模式類似於現在的雲計算，思想是按需分配。

   與粗粒度模式一樣，應用程式啟動時，先會啟動 executor，但每個 executor 佔用資源僅僅是自己運行所需的資源，不需要考慮將來要運行的任務，之後，mesos 會為每個 executor 動態分配資源，每分配一些，便可以運行一個新任務，單個 Task 運行完之後可以馬上釋放對應的資源。

   每個 Task 會彙報狀態給 Mesos slave 和 Mesos Master，便於更加細粒度管理和容錯，這種調度模式類似於 MapReduce 調度模式，每個 Task 完全獨立，優點是便於資源控制和隔離，但缺點也很明顯，短作業運行延遲大。

總結

這三種分散式部署方式各有利弊，通常需要根據實際情況決定採用哪種方案。

進行方案選擇時，往往要考慮公司的技術路線（採用 Hadoop 生態系統還是其他生態系統）、相關技術人才儲備等。上面涉及到 Spark 的許多部署模式，究竟哪種模式好這個很難說，需要根據你的需求，如果你只是測試 Spark Application，你可以選擇 local 模式。而如果你數據量不是很多，Standalone 是個不錯的選擇。當你需要統一管理集群資源（Hadoop、Spark 等），那麼你可以選擇 Yarn 或者 mesos，但是這樣維護成本就會變高。

• 從對比上看，mesos 似乎是 Spark 更好的選擇，也是被官方推薦的
• 但如果你同時運行 hadoop 和 Spark,從兼容性上考慮，Yarn 是更好的選擇。
• 如果你不僅運行了 hadoop，spark。還在資源管理上運行了 docker，Mesos 更加通用。
• Standalone 對於小規模計算集群更適合！

Standalone 模式

Standalone 另可分兩種子模式：

• 單機
• 集群

當然，集群部署的前提是單機的部署完成，根據自己的需求調整即可

單機部署

Spark 雖然是大規模的計算框架，但也支援在單機上運行

修改配置文件

進入 Spark 配置目錄

cd $SPARK_HOME/conf

日誌配置

創建/複製

cp log4j.properties.template log4j.properties

我們修改 log4j.rootCategory 的 「INFO」修改為「WARN」，這一步是修改日誌等級，可避免測試中輸出太多資訊

spark-env.sh

創建/複製

 cp spark-env.sh.template spark-env.sh

添加HOME 變數：JAVA_HOME、SPARK_HOME、SCALA_HOME

考慮我們已經添加至環境變數文件里了，所以我們刷新配置的環境文件即可

spark-env.sh腳本會在啟動 Spark 時載入，內容包含很多配置選項及說明，在以後會用到少部分，感興趣可以仔細閱讀這個文件的注釋內容。

啟動Spark 服務

這一節將啟動 Spark 的 master 主節點和 slave 從節點

也會介紹 spark 單機模式和集群模式的部署區別

啟動主節點

前往sbin 目錄

cd $SPARK_HOME/sbin

啟動

./start-master.sh

沒有報錯的話表示 master 已經啟動成功

master 默認可以通過 web 訪問//localhost:8080

圖中所示，master 中暫時還沒有一個 worker ，我們啟動 worker 時需要 master 的參數，該參數已經在上圖中標誌出來：spark://master:7077，請在執行後續命令時替換成你自己的參數。

啟動從節點

啟動 slave

./start-slave.sh spark://master:7077

沒有報錯表示啟動成功，再次刷新瀏覽器頁面可以看到下圖所示新的 worker 已經添加

也可以用jps命令查看啟動的服務，應該會列出Master和Worker。

測試實例

使用 spark-shell 連接 master ，注意把 MASTER 參數替換成你實驗環境中的實際參數

MASTER=spark://master:7077 spark-shell

刷新 master 的 web 頁面，可以看到新的Running Applications，如下圖所示：

當退出 spark-shell 時，這個 application 會移動到Completed Applications一欄。

可以自己點擊頁面中的 Application 和 Workers 的鏈接查看並了解相關資訊。

停止服務

停止服務的腳本為./sbin/stop-all.sh。

./stop-all.sh

但我建議依次關閉

./stop-master.sh
./stop-slave.sh

通過 jps 可以看到，master 與 worker 進程都已經停止

集群部署

修改配置

在 「單機模式」小節下的「修改配置文件」的基礎上進行添加/修改

進入 Spark 配置目錄

cd $SPARK_HOME/conf

spark-env.sh

參數解讀：

• SPARK_MASTER_HOST = Master的主機名
• SPARK_MASTER_PORT = 提交Application的埠，默認7077，可更改
• SPARK_WORKER_CORES = 每一個Worker最多可以使用的cpu核個數
• SPARK_WORKER_MEMORY = 每個Worker最多可以使用的記憶體

其實你完全可以參考spark-env.sh內的注釋

編輯

vim spark-env.sh

做出如下修改（位置非固定）：

slaves

修改 slaves 配置文件，添加 Worker 的主機列表

複製/創建

cp slaves.template slaves

修改localhost 為你需要的機器的 HostName

你可以參考我的：

注意需要先把所有主機名輸入到 /etc/hosts 避免無法解析

同步

此處不進行贅述，具體操作大家自行搜索，或者查看我關於 「HADOOP部署」的相關文章

大致操作如下：

• ssh-keygen 命令配合 ssh-copy-id 命令實現ssh免密
• scp 命令同步所有設置（指Spark 下的 conf 文件下，或者同步 Spark 文件）

啟動集群

前往 master 機器下執行

前往sbin 目錄

cd $SPARK_HOME/sbin

在這台機啟動集群

./start-all.sh

啟動的步驟和「單機部署」下的「啟動 Spark 服務」一致，關閉也一致

start-all.sh 和 start-master.sh、start-slave.sh 和 Hadoop 里的 start-all.sh、start-yarn.sh、start-dfs.sh 的關係大致一樣

Web：

Spark 互動式執行

Spark-Shell

Spark-Shell是 Spark 自帶的一個 Scala 交互 Shell ，可以以腳本方式進行互動式執行，類似直接用 Python 及其他腳本語言的 Shell 。

進入Spark-Shell只需要執行spark-shell即可：

spark-shell

（前提是你配置好了 Spark 的環境變數）

進入到Spark-Shell後可以使用Ctrl D組合鍵退出 Shell。

在Spark-Shell中我們可以使用 Scala 的語法進行簡單的測試，比如我們運行下面幾個語句獲得文件/etc/protocols的行數以及第一行的內容：

var f = sc.textFile("/etc/protocols")
f.count()
f.first()

上面的操作中創建了一個 RDD file，執行了兩個簡單的操作：

• count()獲取 RDD 的行數
• first()獲取第一行的內容

我們繼續執行其他操作，比如查找有多少行含有tcp和udp字元串：

f.filter(line => line.contains("tcp")).count()
f.filter(line => line.contains("udp")).count()

查看一共有多少個不同單詞的方法，這裡用到 Mapreduce 的思路：

var wordcount = file.flatMap(line => line.split(" ")).map(word => (word,1)).reduceByKey(_+_)

wordcount.count()

上面兩步驟我們發現，/etc/protocols中各有一行含有tcp與udp字元串，並且一共有 442 個不同的單詞。

上面每個語句的具體含義這裡不展開，可以結合你閱讀的文章進行理解，這裡僅僅提供一個簡單的例子讓大家對 Spark 運算有基本認識。

操作完成後，Ctrl D組合鍵退出 Shell。

Pyspark

Pyspark 類似 Spark-Shell ，是一個 Python 的交互 Shell 。

執行pyspark啟動進入 Pyspark：

pyspark

退出方法仍然是Ctrl D組合鍵。

在 Pyspark 中，我們可以用 Python 語法執行 Spark-Shell 中的操作，比如下面的語句獲得文件/etc/protocols 的行數以及第一行的內容：

file = sc.textFile("/etc/protocols")
file.count()
file.first()

操作完成後，Ctrl D組合鍵退出 Shell。

對於 Pyspark 大家可以自行學習拓展，可以參考官方文檔Spark Python API

提交應用程式

在Spark bin目錄下的spark-submit可以用來在集群上啟動應用程式。它可以通過統一的介面使用Spark支援的所有集群管理器 ，所有你不必為每一個管理器做相應的配置。

用spark-submit啟動應用程式

bin/spark-submit腳本負責建立包含Spark以及其依賴的類路徑（classpath），它支援不同的集群管理器以及Spark支援的載入模式。

./bin/spark-submit \
  --class <main-class>
  --master <master-url> \
  --deploy-mode <deploy-mode> \
  --conf <key>=<value> \
  ... # other options
  <application-jar> \
  [application-arguments]

一些常用的選項是：

• --class：你的應用程式的入口點(如org.apache.spark.examples.SparkPi)
• --master：集群的master URL(如spark://23.195.26.187:7077)
• --deploy-mode：在worker節點部署你的driver(cluster)或者本地作為外部客戶端（client）。默認是client。
• --conf：任意的Spark配置屬性，格式是key=value。
• application-jar：包含應用程式以及其依賴的jar包的路徑。這個URL必須在集群中全局可見，例如，存在於所有節點的hdfs://路徑或file://路徑
• application-arguments：傳遞給主類的主方法的參數

一個通用的部署策略是從網關集群提交你的應用程式，這個網關機器和你的worker集群物理上協作。在這種設置下，client模式是適合的。在client模式下，driver直接在spark-submit進程 中啟動，而這個進程直接作為集群的客戶端。應用程式的輸入和輸出都和控制台相連接。因此，這種模式特別適合涉及REPL的應用程式。

另一種選擇，如果你的應用程式從一個和worker機器相距很遠的機器上提交，通常情況下用cluster模式減少drivers和executors的網路遲延。注意，cluster模式目前不支援獨立集群、 mesos集群以及python應用程式。

有幾個我們使用的集群管理器特有的可用選項。例如，在Spark獨立集群的cluster模式下，你也可以指定--supervise用來確保driver自動重啟（如果它因為非零退出碼失敗）。 為了列舉spark-submit所有的可用選項，用--help運行它。

# Run application locally on 8 cores
./bin/spark-submit \
  --class org.apache.spark.examples.SparkPi \
  --master local[8] \
  /path/to/examples.jar \
  100

# Run on a Spark Standalone cluster in client deploy mode
./bin/spark-submit \
  --class org.apache.spark.examples.SparkPi \
  --master spark://master:7077 \
  --executor-memory 20G \
  --total-executor-cores 100 \
  /path/to/examples.jar \
  1000

# Run on a Spark Standalone cluster in cluster deploy mode with supervise
./bin/spark-submit \
  --class org.apache.spark.examples.SparkPi \
  --master spark://master:7077 \
  --deploy-mode cluster
  --supervise
  --executor-memory 20G \
  --total-executor-cores 100 \
  /path/to/examples.jar \
  1000

# Run on a YARN cluster
export HADOOP_CONF_DIR=XXX
./bin/spark-submit \
  --class org.apache.spark.examples.SparkPi \
  --master yarn-cluster \  # can also be `yarn-client` for client mode
  --executor-memory 20G \
  --num-executors 50 \
  /path/to/examples.jar \
  1000

# Run a Python application on a Spark Standalone cluster
./bin/spark-submit \
  --master spark://master:7077 \
  examples/src/main/python/pi.py \
  1000

Master URLs

傳遞給Spark的url可以用下面的模式

Spark Standalone擴展

手動啟動集群

你能夠通過下面的方式啟動獨立的master伺服器。

./sbin/start-master.sh

一旦啟動，master將會為自己列印出spark://HOST:PORT URL，你能夠用它連接到workers或者作為"master"參數傳遞給SparkContext。你也可以在master web UI上發現這個URL， master web UI默認的地址是//localhost:8080。

相同的，你也可以啟動一個或者多個workers或者將它們連接到master。

./bin/spark-class org.apache.spark.deploy.worker.Worker spark://IP:PORT

一旦你啟動了一個worker，查看master web UI。你可以看到新的節點列表以及節點的CPU數以及記憶體。

下面的配置參數可以傳遞給master和worker。

集群啟動腳本

為了用啟動腳本啟動Spark獨立集群，你應該在你的Spark目錄下建立一個名為conf/slaves的文件，這個文件必須包含所有你要啟動的Spark worker所在機器的主機名，一行一個。如果 conf/slaves不存在，啟動腳本默認為單個機器（localhost），這台機器對於測試是有用的。注意，master機器通過ssh訪問所有的worker。在默認情況下，SSH是並行運行，需要設置無密碼（採用私有密鑰）的訪問。 如果你沒有設置為無密碼訪問，你可以設置環境變數SPARK_SSH_FOREGROUND，為每個worker提供密碼。

一旦你設置了這個文件，你就可以通過下面的shell腳本啟動或者停止你的集群。

• sbin/start-master.sh：在機器上啟動一個master實例
• sbin/start-slaves.sh：在每台機器上啟動一個slave實例
• sbin/start-all.sh：同時啟動一個master實例和所有slave實例
• sbin/stop-master.sh：停止master實例
• sbin/stop-slaves.sh：停止所有slave實例
• sbin/stop-all.sh：停止master實例和所有slave實例

注意，這些腳本必須在你的Spark master運行的機器上執行，而不是在你的本地機器上面。

你可以在conf/spark-env.sh中設置環境變數進一步配置集群。利用conf/spark-env.sh.template創建這個文件，然後將它複製到所有的worker機器上使設置有效。下面的設置可以起作用：

注意，啟動腳本還不支援windows。為了在windows上啟動Spark集群，需要手動啟動master和workers。

SPARK_MASTER_OPTS支援一下的系統屬性：

SPARK_WORKER_OPTS支援的系統屬性：

連接一個應用程式到集群中

為了在Spark集群中運行一個應用程式，簡單地傳遞spark://IP:PORT URL到SparkContext

為了在集群上運行一個互動式的Spark shell，運行一下命令：

./bin/spark-shell --master spark://IP:PORT

你也可以傳遞一個選項--total-executor-cores <numcores>去控制spark-shell的核數。

啟動Spark應用程式

spark-submit腳本支援最直接的提交一個Spark應用程式到集群。對於獨立部署的集群，Spark目前支援兩種部署模式。在client模式中，driver啟動進程與 客戶端提交應用程式所在的進程是同一個進程。然而，在cluster模式中，driver在集群的某個worker進程中啟動，只有客戶端進程完成了提交任務，它不會等到應用程式完成就會退出。

如果你的應用程式通過Spark submit啟動，你的應用程式jar包將會自動分發到所有的worker節點。對於你的應用程式依賴的其它jar包，你應該用--jars符號指定（如--jars jar1,jar2）。

另外，cluster模式支援自動的重啟你的應用程式（如果程式一非零的退出碼退出）。為了用這個特徵，當啟動應用程式時，你可以傳遞--supervise符號到spark-submit。如果你想殺死反覆失敗的應用， 你可以通過如下的方式：

./bin/spark-class org.apache.spark.deploy.Client kill <master url=""> <driver id="">

你可以在獨立部署的Master web UI（//:8080）中找到driver ID。

資源調度

獨立部署的集群模式僅僅支援簡單的FIFO調度器。然而，為了允許多個並行的用戶，你能夠控制每個應用程式能用的最大資源數。在默認情況下，它將獲得集群的所有核，這隻有在某一時刻只 允許一個應用程式才有意義。你可以通過spark.cores.max在SparkConf中設置核數。

val conf = new SparkConf()
             .setMaster(...)
             .setAppName(...)
             .set("spark.cores.max", "10")
val sc = new SparkContext(conf)

另外，你可以在集群的master進程中配置spark.deploy.defaultCores來改變默認的值。在conf/spark-env.sh添加下面的行：

export SPARK_MASTER_OPTS="-Dspark.deploy.defaultCores=<value>"

這在用戶沒有配置最大核數的共享集群中是有用的。

高可用

默認情況下，獨立的調度集群對worker失敗是有彈性的（在Spark本身的範圍內是有彈性的，對丟失的工作通過轉移它到另外的worker來解決）。然而，調度器通過master去執行調度決定， 這會造成單點故障：如果master死了，新的應用程式就無法創建。為了避免這個，我們有兩個高可用的模式。

用ZooKeeper的備用master

利用ZooKeeper去支援領導選舉以及一些狀態存儲，你能夠在你的集群中啟動多個master，這些master連接到同一個ZooKeeper實例上。一個被選為「領導」，其它的保持備用模式。如果當前 的領導死了，另一個master將會被選中，恢復老master的狀態，然後恢復調度。整個的恢復過程大概需要1到2分鐘。注意，這個恢復時間僅僅會影響調度新的應用程式-運行在失敗master中的 應用程式不受影響。

配置

為了開啟這個恢復模式，你可以用下面的屬性在spark-env中設置SPARK_DAEMON_JAVA_OPTS。

可能的陷阱：如果你在集群中有多個masters，但是沒有用zookeeper正確的配置這些masters，這些masters不會發現彼此，會認為它們都是leaders。這將會造成一個不健康的集群狀態（因為所有的master都會獨立的調度）。

細節

zookeeper集群啟動之後，開啟高可用是簡單的。在相同的zookeeper配置（zookeeper URL和目錄）下，在不同的節點上簡單地啟動多個master進程。master可以隨時添加和刪除。

為了調度新的應用程式或者添加worker到集群，它需要知道當前leader的IP地址。這可以通過簡單的傳遞一個master列表來完成。例如，你可能啟動你的SparkContext指向spark://host1:port1,host2:port2。 這將造成你的SparkContext同時註冊這兩個master-如果host1死了，這個配置文件將一直是正確的，因為我們將找到新的leader-host2。

"registering with a Master"和正常操作之間有重要的區別。當啟動時，一個應用程式或者worker需要能夠發現和註冊當前的leader master。一旦它成功註冊，它就在系統中了。如果 錯誤發生，新的leader將會接觸所有之前註冊的應用程式和worker，通知他們領導關係的變化，所以它們甚至不需要事先知道新啟動的leader的存在。

由於這個屬性的存在，新的master可以在任何時候創建。你唯一需要擔心的問題是新的應用程式和workers能夠發現它並將它註冊進來以防它成為leader master。

用本地文件系統做單節點恢復

zookeeper是生產環境下最好的選擇，但是如果你想在master死掉後重啟它，FILESYSTEM模式可以解決。當應用程式和worker註冊，它們擁有足夠的狀態寫入提供的目錄，以至於在重啟master 進程時它們能夠恢復。

配置

為了開啟這個恢復模式，你可以用下面的屬性在spark-env中設置SPARK_DAEMON_JAVA_OPTS。

細節

• 這個解決方案可以和監控器/管理器（如monit）相配合，或者僅僅通過重啟開啟手動恢復。
• 雖然文件系統的恢復似乎比沒有做任何恢復要好，但對於特定的開發或實驗目的，這種模式可能是次優的。特別是，通過stop-master.sh殺掉master不會清除它的恢復狀態，所以，不管你何時啟動一個新的master，它都將進入恢復模式。這可能使啟動時間增加到1分鐘。
• 雖然它不是官方支援的方式，你也可以創建一個NFS目錄作為恢複目錄。如果原始的master節點完全死掉，你可以在不同的節點啟動master，它可以正確的恢復之前註冊的所有應用程式和workers。未來的應用程式會發現這個新的master。

在YARN上運行Spark

配置

大部分為Spark on YARN模式提供的配置與其它部署模式提供的配置相同。下面這些是為Spark on YARN模式提供的配置。

Hadoop 配置

記得到yarn-site-xml下添加：

vim  $HADOOP_HOME/etc/hadoop/yarn-site.xml

<property>
    <name>yarn.nodemanager.pmem-check-enabled</name>
    <value>false</value>
</property>

<property>
    <name>yarn.nodemanager.vmem-check-enabled</name>
    <value>false</value>
</property> 

Spark屬性

在YARN上啟動Spark

確保HADOOP_CONF_DIR或YARN_CONF_DIR指向的目錄包含Hadoop集群的（客戶端）配置文件。這些配置用於寫數據到dfs和連接到YARN ResourceManager。

有兩種部署模式可以用來在YARN上啟動Spark應用程式。在yarn-cluster模式下，Spark driver運行在application master進程中，這個進程被集群中的YARN所管理，客戶端會在初始化應用程式 之後關閉。在yarn-client模式下，driver運行在客戶端進程中，application master僅僅用來向YARN請求資源。

和Spark單獨模式以及Mesos模式不同，在這些模式中，master的地址由"master"參數指定，而在YARN模式下，ResourceManager的地址從Hadoop配置得到。因此master參數是簡單的yarn-client和yarn-cluster。

在yarn-cluster模式下啟動Spark應用程式。

./bin/spark-submit --class path.to.your.Class --master yarn-cluster [options] <app jar=""> [app options]

例子：

$ ./bin/spark-submit --class org.apache.spark.examples.SparkPi \
    --master yarn-cluster \
    --num-executors 3 \
    --driver-memory 4g \
    --executor-memory 2g \
    --executor-cores 1 \
    --queue thequeue \
    lib/spark-examples*.jar \
    10

以上啟動了一個YARN客戶端程式用來啟動默認的 Application Master，然後SparkPi會作為Application Master的子執行緒運行。客戶端會定期的輪詢Application Master用於狀態更新並將 更新顯示在控制台上。一旦你的應用程式運行完畢，客戶端就會退出。

在yarn-client模式下啟動Spark應用程式，運行下面的shell腳本

$ ./bin/spark-shell --master yarn-client

添加其它的jar

在yarn-cluster模式下，driver運行在不同的機器上，所以離開了保存在本地客戶端的文件，SparkContext.addJar將不會工作。為了使SparkContext.addJar用到保存在客戶端的文件， 在啟動命令中加上--jars選項。

$ ./bin/spark-submit --class my.main.Class \
    --master yarn-cluster \
    --jars my-other-jar.jar,my-other-other-jar.jar
    my-main-jar.jar
    app_arg1 app_arg2

注意事項

• 在Hadoop 2.2之前，YARN不支援容器核的資源請求。因此，當運行早期的版本時，通過命令行參數指定的核的數量無法傳遞給YARN。在調度決策中，核請求是否兌現取決於用哪個調度器以及 如何配置調度器。
• Spark executors使用的本地目錄將會是YARN配置（yarn.nodemanager.local-dirs）的本地目錄。如果用戶指定了spark.local.dir，它將被忽略。
• --files和--archives選項支援指定帶 # 號文件名。例如，你能夠指定--files localtest.txt#appSees.txt，它上傳你在本地命名為localtest.txt的文件到HDFS，但是將會鏈接為名稱appSees.txt。當你的應用程式運行在YARN上時，你應該使用appSees.txt去引用該文件。
• 如果你在yarn-cluster模式下運行SparkContext.addJar，並且用到了本地文件， --jars選項允許SparkContext.addJar函數能夠工作。如果你正在使用 HDFS, HTTP, HTTPS或FTP，你不需要用到該選項

Spark配置

Spark提供三個位置用來配置系統：

• Spark properties控制大部分的應用程式參數，可以用SparkConf對象或者java系統屬性設置
• Environment variables可以通過每個節點的conf/spark-env.sh腳本設置每台機器的設置。例如IP地址
• Logging可以通過log4j.properties配置

Spark屬性

Spark屬性控制大部分的應用程式設置，並且為每個應用程式分別配置它。這些屬性可以直接在SparkConf上配置，然後傳遞給SparkContext。SparkConf 允許你配置一些通用的屬性（如master URL、應用程式明）以及通過set()方法設置的任意鍵值對。例如，我們可以用如下方式創建一個擁有兩個執行緒的應用程式。注意，我們用local[2]運行，這意味著兩個執行緒-表示最小的 並行度，它可以幫助我們檢測當在分散式環境下運行的時才出現的錯誤。

val conf = new SparkConf()
             .setMaster("local[2]")
             .setAppName("CountingSheep")
             .set("spark.executor.memory", "1g")
val sc = new SparkContext(conf)

注意，我們在本地模式中擁有超過1個執行緒。和Spark Streaming的情況一樣，我們可能需要一個執行緒防止任何形式的飢餓問題。

動態載入Spark屬性

在一些情況下，你可能想在SparkConf中避免硬編碼確定的配置。例如，你想用不同的master或者不同的記憶體數運行相同的應用程式。Spark允許你簡單地創建一個空conf。

val sc = new SparkContext(new SparkConf())

然後你在運行時提供值。

./bin/spark-submit --name "My app" --master local[4] --conf spark.shuffle.spill=false
  --conf "spark.executor.extraJavaOptions=-XX:+PrintGCDetails -XX:+PrintGCTimeStamps" myApp.jar

Spark shell和spark-submit工具支援兩種方式動態載入配置。第一種方式是命令行選項，例如--master，如上面shell顯示的那樣。spark-submit可以接受任何Spark屬性，用--conf 標記表示。但是那些參與Spark應用程式啟動的屬性要用特定的標記表示。運行./bin/spark-submit --help將會顯示選項的整個列表。

bin/spark-submit也會從conf/spark-defaults.conf中讀取配置選項，這個配置文件中，每一行都包含一對以空格分開的鍵和值。例如：

spark.master            spark://5.6.7.8:7077
spark.executor.memory   512m
spark.eventLog.enabled  true
spark.serializer        org.apache.spark.serializer.KryoSerializer

任何標籤（flags）指定的值或者在配置文件中的值將會傳遞給應用程式，並且通過SparkConf合併這些值。在SparkConf上設置的屬性具有最高的優先順序，其次是傳遞給spark-submit 或者spark-shell的屬性值，最後是spark-defaults.conf文件中的屬性值。

查看Spark屬性

在//<driver>:4040上的應用程式web UI在「Environment」標籤中列出了所有的Spark屬性。這對你確保設置的屬性的正確性是很有用的。注意，只有通過spark-defaults.conf, SparkConf以及 命令行直接指定的值才會顯示。對於其它的配置屬性，你可以認為程式用到了默認的值。

可用的屬性

控制內部設置的大部分屬性都有合理的默認值，一些最通用的選項設置如下：

應用程式屬性

運行環境

Shuffle行為(Behavior)

Spark UI

壓縮和序列化

Networking

Security

Spark Streaming

環境變數

通過環境變數配置確定的Spark設置。環境變數從Spark安裝目錄下的conf/spark-env.sh腳本讀取（或者windows的conf/spark-env.cmd）。在獨立的或者Mesos模式下，這個文件可以給機器 確定的資訊，如主機名。當運行本地應用程式或者提交腳本時，它也起作用。

注意，當Spark安裝時，conf/spark-env.sh默認是不存在的。你可以複製conf/spark-env.sh.template創建它。

可以在spark-env.sh中設置如下變數：

除了以上這些，Spark standalone cluster scripts也可以設置一些選項。例如 每台機器使用的核數以及最大記憶體。

因為spark-env.sh是shell腳本，其中的一些可以以編程方式設置。例如，你可以通過特定的網路介面計算SPARK_LOCAL_IP。

配置Logging

Spark用log4j logging。你可以通過在conf目錄下添加log4j.properties文件來配置。一種方法是複製log4j.properties.template文件。

Spark調優

由於大部分Spark計算都是在記憶體中完成的，所以Spark程式的瓶頸可能由集群中任意一種資源導致，如：CPU、網路頻寬、或者記憶體等。最常見的情況是，數據能裝進記憶體，而瓶頸是網路頻寬；當然，有時候我們也需要做一些優化調整來減少記憶體佔用，例如將RDD以序列化格式保存。 本文將主要涵蓋兩個主題：1.數據序列化（這對於優化網路性能極為重要）；2.減少記憶體佔用以及記憶體調優。同時，我們也會提及其他幾個比較小的主題。

數據序列化

序列化在任何一種分散式應用性能優化時都扮演幾位重要的角色。如果序列化格式序列化過程緩慢，或者需要佔用位元組很多，都會大大拖慢整體的計算效率。 通常，序列化都是Spark應用優化時首先需要關注的地方。Spark著眼於便利性（允許你在計算過程中使用任何Java類型）和性能的一個平衡。Spark主要提供了兩個序列化庫：

• Java serialization:默認情況，Spark使用Java自帶的ObjectOutputStream框架來序列化對象，這樣任何實現了 java.io.Serializable 介面的對象，都能被序列化。同時，你還可以通過擴展java.io.Externalizable 來控制序列化性能。Java序列化很靈活但性能較差，同時序列化後佔用的位元組數也較多。
• Kryo serialization: Spark還可以使用Kryo庫（版本2）提供更高效的序列化格式。Kryo的序列化速度和位元組佔用都比Java序列化好很多（通常是10倍左右），但Kryo不支援所有實現了Serializable 介面的類型，它需要你在程式中register 需要序列化的類型，以得到最佳性能。

要切換使用 Kryo，你可以在 SparkConf 初始化的時候調用 conf.set(「spark.serializer」, 「org.apache.spark.serializer.KryoSerializer」)。這個設置不僅控制各個worker節點之間的混洗數據序列化格式，同時還控制RDD存到磁碟上的序列化格式。 目前，Kryo不是默認的序列化格式，因為它需要你在使用前註冊需要序列化的類型，不過我們還是建議在對網路敏感的應用場景下使用Kryo。

Spark對一些常用的Scala核心類型,如在Twitter chill 庫的AllScalaRegistrar中，自動使用Kryo序列化格式。

如果你的自定義類型需要使用Kryo序列化，可以用 registerKryoClasses 方法先註冊：

val conf = new SparkConf().setMaster(...).setAppName(...)
conf.registerKryoClasses(Array(classOf[MyClass1], classOf[MyClass2]))
val sc = new SparkContext(conf)

Kryo的文檔中有詳細描述了更多的高級選項，如：自定義序列化程式碼等。

如果你的對象很大，你可能需要增大 spark.kryoserializer.buffer 配置項。其值至少需要大於最大對象的序列化長度。

最後，如果你不註冊需要序列化的自定義類型，Kryo也能工作，不過每一個對象實例的序列化結果都會包含一份完整的類名，這有點浪費空間。

記憶體調優

記憶體佔用調優主要需要考慮3點：數據佔用的總記憶體（你會希望整個數據集都能裝進記憶體）；訪問數據集中每個對象的開銷；垃圾回收的開銷（如果你的數據集中對象周轉速度很快的話）。

一般情況下，Java對象的訪問時很快的，但同時Java對象會比原始數據（僅包含各個欄位值）佔用的空間多2~5倍。主要原因有：

• 每個Java對象都有一個對象頭（object header），對象頭大約佔用16位元組，其中包含像其對應class的指針這樣的資訊。對於一些包含較少數據的對象（比如只包含一個Int欄位），這個對象頭可能比對象數據本身還大。
• Java字元串（String）有大約40位元組額外開銷（Java String以Char數據的形式保存原始數據，所以需要一些額外的欄位，如數組長度等），並且每個字元都以兩位元組的UTF-16編碼在內部保存。因此，10個字元的String很容易就佔了60位元組。
• 一些常見的集合類，如 HashMap、LinkedList，使用的是鏈表類數據結構，因此它們對每項數據都有一個包裝器。這些包裝器對象不僅其自身就有「對象頭」，同時還有指向下一個包裝器對象的鏈表指針（通常為8位元組）。
• 原始類型的集合通常也是以「裝箱」的形式包裝成對象（如：java.lang.Integer）。

本節只是Spark記憶體管理的一個概要，下面我們會更詳細地討論各種Spark記憶體調優的具體策略。特別地，我們會討論如何評估數據的記憶體使用量，以及如何改進 – 要麼改變你的數據結構，要麼以某種序列化格式存儲數據。 最後，我們還會討論如何調整Spark的快取大小，以及如何調優Java的垃圾回收器。

記憶體管理概覽

Spark中記憶體主要用於兩類目的：執行計算和數據存儲。執行計算的記憶體主要用於Shuffle、關聯（join）、排序（sort）以及聚合（aggregation），而數據存儲的記憶體主要用於快取和集群內部數據傳播。Spark中執行計算和數據存儲都是共享同一個記憶體區域（M）。 如果執行計算沒有佔用記憶體，那麼數據存儲可以申請佔用所有可用的記憶體，反之亦然。執行計算可能會搶佔數據存儲使用的記憶體，並將存儲於記憶體的數據逐出記憶體，直到數據存儲佔用的記憶體比例降低到一個指定的比例（R）。 換句話說，R是M基礎上的一個子區域，這個區域的記憶體數據永遠不會被逐出記憶體。然而，數據存儲不會搶佔執行計算的記憶體。

這樣設計主要有這麼幾個需要考慮的點。首先，不需要快取數據的應用可以把整個空間用來執行計算，從而避免頻繁地把數據吐到磁碟上。其次，需要快取數據的應用能夠有一個數據存儲比例（R）的最低保證，也避免這部分快取數據被全部逐出記憶體。最後，這個實現方式能夠在默認情況下，為大多數使用場景提供合理的性能，而不需要專家級用戶來設置記憶體使用如何劃分。

雖然有兩個記憶體劃分相關的配置參數，但一般來說，用戶不需要設置，因為默認值已經能夠適用於絕大部分的使用場景：

• spark.memory.fraction:表示上面M的大小，其值為相對於JVM堆記憶體的比例（默認0.75）。剩餘的25%是為其他用戶的數據結構、Spark內部元數據以及避免OOM錯誤的安全預留空間。
• spark.memory.storageFraction:表示上面R的大小，其值為相對於M的一個比例（默認0.5）。R是M中專門用於快取數據塊的部分，這部分數據塊永遠不會因執行計算任務而逐出記憶體。

評估記憶體消耗

確定一個數據集佔用記憶體總量最好的辦法就是，創建一個RDD，並快取到記憶體中，然後再到web UI上」Storage」頁面查看。頁面上會展示這個RDD總共佔用了多少記憶體。

要評估一個特定對象的記憶體佔用量，可以用 SizeEstimator.estimate 方法。這個方法對試驗哪種數據結構能夠裁剪記憶體佔用量比較有用，同時，也可以幫助用戶了解廣播變數在每個執行器堆上佔用的記憶體量。

數據結構調優

減少記憶體消耗的首要方法就是避免過多的Java封裝（減少對象頭和額外輔助欄位），比如基於指針的數據結構和包裝對象等。以下有幾條建議：

• 設計數據結構的時候，優先使用對象數組和原生類型，減少對複雜集合類型（如：HashMap）的使用。fastutil 提供了一些很方便的原生類型集合，同時兼容Java標準庫。
• 儘可能避免嵌套大量的小對象和指針。
• 對應鍵值應盡量使用數值型或枚舉型，而不是字元串型。
• 如果記憶體小於32GB，可以設置JVM標誌參數 -XX:+UseCompressdOops 將指針設為4位元組而不是8位元組。你可以在 spark-env.sh 中設置這個參數。

序列化RDD存儲

如果經過上面的調整後，存儲的數據對象還是太大，那麼你可以試試將這些對象以序列化格式存儲，所需要做的只是通過 RDD persistence API 設置好存儲級別，如：MEMORY_ONLY_SER。Spark會將RDD的每個分區以一個巨大的位元組數組形式存儲起來。以序列化格式存儲的唯一缺點就是訪問數據會變慢一點，因為Spark需要反序列化每個被訪問的對象。 如果你需要序列化快取數據，我們強烈建議你使用Kryo，和Java序列化相比，Kryo能大大減少序列化對象佔用的空間（當然也比原始Java對象小很多）。

垃圾回收調優

JVM的垃圾回收在某些情況下可能會造成瓶頸，比如，你的RDD存儲經常需要「換入換出」（新RDD搶佔了老RDD記憶體，不過如果你的程式沒有這種情況的話那JVM垃圾回收一般不是問題，比如，你的RDD只是載入一次，後續只是在這一個RDD上做操作）。當Java需要把老對象逐出記憶體的時候，JVM需要跟蹤所有的Java對象，並找出哪些對象已經沒有用了。 概括起來就是，垃圾回收的開銷和對象個數成正比，所以減少對象的個數（比如用Int數組取代LinkedList），就能大大減少垃圾回收的開銷。 當然，一個更好的方法就如前面所說的，以序列化形式存儲數據，這時每個RDD分區都只包含有一個對象了（一個巨大的位元組數組）。在嘗試其他技術方案前，首先可以試試用序列化RDD的方式（serialized caching）評估一下GC是不是一個瓶頸。

如果你的作業中各個任務需要的工作記憶體和節點上存儲的RDD快取佔用的記憶體產生衝突，那麼GC很可能會出現問題。下面我們將討論一下如何控制好RDD快取使用的記憶體空間，以減少這種衝突。

衡量GC的影響

GC調優的第一步是統計一下，垃圾回收啟動的頻率以及GC所使用的總時間。給JVM設置一下這幾個參數（參考Spark配置指南，查看Spark作業中的Java選項參數）：-verbose:gc -XX:+PrintGCDetails，就可以在後續Spark作業的worker日誌中看到每次GC花費的時間。 注意，這些日誌是在集群worker節點上（在各節點的工作目錄下stdout文件中），而不是你的驅動器所在節點。

高級GC調優

為了進一步調優GC，我們就需要對JVM記憶體管理有一個基本的了解：

• Java堆記憶體可分配的空間有兩個區域：新生代（Young generation）和老年代（Old generation）。新生代用以保存生存周期短的對象，而老年代則是保存生存周期長的對象。
• 新生代區域被進一步劃分為三個子區域：Eden，Survivor1，Survivor2。
• 簡要描述一下垃圾回收的過程：如果Eden區滿了，則啟動一輪minor GC回收Eden中的對象，生存下來（沒有被回收掉）的Eden中的對象和Survivor1區中的對象一併複製到Survivor2中。 兩個Survivor區域是互相切換使用的（就是說，下次從Eden和Survivor2中複製到Survivor1中）。如果某個對象的年齡（每次GC所有生存下來的對象長一歲）超過某個閾值，或者Survivor2（下次是Survivor1）區域滿了，則將對象移到老年代（Old區）。最終如果老生代也滿了，就會啟動full GC。

Spark GC調優的目標就是確保老年代（Old generation ）只保存長生命周期RDD，而同時新生代（Young generation）的空間又能足夠保存短生命周期的對象。這樣就能在任務執行期間，避免啟動full GC。以下是GC調優的主要步驟：

• 從GC的統計日誌中觀察GC是否啟動太多。如果某個任務結束前，多次啟動了full GC，則意味著用以執行該任務的記憶體不夠。
• 如果GC統計資訊中顯示，老生代記憶體空間已經接近存滿，可以通過降低spark.memory.storageFraction 來減少RDD快取佔用的記憶體；減少快取對象總比任務執行緩慢要強！
• 如果major GC比較少，但minor GC很多的話，可以多分配一些Eden記憶體。你可以把Eden的大小設為高於各個任務執行所需的工作記憶體。如果要把Eden大小設為E，則可以這樣設置新生代區域大小：-Xmn=4/3*E。（放大4/3倍，主要是為了給Survivor區域保留空間）
• 舉例來說，如果你的任務會從HDFS上讀取數據，那麼單個任務的記憶體需求可以用其所讀取的HDFS數據塊的大小來評估。需要特別注意的是，解壓後的HDFS塊是解壓前的2~3倍。所以如果我們希望保留3~4個任務並行的工作記憶體，並且HDFS塊大小為64MB，那麼可以評估Eden的大小應該設為 4*3*64MB。
• 最後，再觀察一下垃圾回收的啟動頻率和總耗時有沒有什麼變化。

我們的很多經驗表明，GC調優的效果和你的程式程式碼以及可用的總記憶體相關。網上還有不少調優的選擇，但總體來說，就是控制好full GC的啟動頻率，就能有效減少垃圾回收開銷。

其他事項

並行度

一般來說集群並不會滿負荷運轉，除非你把每個操作的並行度都設得足夠大。Spark會自動根據對應的輸入文件大小來設置「map」類運算元的並行度（當然你可以通過一個SparkContext.textFile等函數的可選參數來控制並行度），而對於想 groupByKey 或reduceByKey這類 「reduce」 運算元，會使用其各父RDD分區數的最大值。你可以將並行度作為構建RDD第二個參數（參考spark.PairRDDFunctions），或者設置 spark.default.parallelism 這個默認值。一般來說，評估並行度的時候，我們建議2~3個任務共享一個CPU。

Reduce任務的記憶體佔用

如果RDD比記憶體要大，有時候你可能收到一個OutOfMemoryError錯誤，其實這是因為你的任務集中的某個任務太大了，如reduce任務groupByKey。Spark的Shuffle運算元（sortByKey，groupByKey，reduceByKey，join等）會在每個任務中構建一個哈希表，以便在任務中對數據分組，這個哈希表有時會很大。最簡單的修復辦法就是增大並行度，以減小單個任務的輸入集。Spark對於200ms以內的短任務支援非常好，因為Spark可以跨任務復用執行器JVM，任務的啟動開銷很小，因此把並行度增加到比集群中總CPU核數沒有任何問題。

廣播大變數

使用SparkContext中的廣播變數相關功能（broadcast functionality）能大大減少每個任務本身序列化的大小，以及集群中啟動作業的開銷。如果你的Spark任務正在使用驅動程式中定義的巨大對象（比如：靜態查詢表），請考慮使用廣播變數替代。Spark會在master上將各個任務的序列化後大小列印出來，所以你可以檢查一下各個任務是否過大；通常來說，大於20KB的任務就值得優化一下。

數據本地性

數據本地性對Spark作業往往會有較大的影響。如果程式碼和其所操作的數據在同一節點上，那麼計算速度肯定會更快一些。但如果二者不在一起，那必然需要移動其中之一。一般來說，移動序列化好的程式碼肯定比挪動一大堆數據要快。Spark就是基於這個一般性原則來構建數據本地性的調度。

數據本地性是指程式碼和其所處理的數據的距離。基於數據當前的位置，數據本地性可以劃分成以下幾個層次（按從近到遠排序）：

• PROCESS_LOCAL 數據和運行的程式碼處於同一個JVM進程內。
• NODE_LOCAL 數據和程式碼處於同一節點。例如，數據處於HDFS上某個節點，而對應的執行器（executor）也在同一個機器節點上。這會比PROCESS_LOCAL稍微慢一些，因為數據需要跨進程傳遞。
• NO_PREF 數據在任何地方處理都一樣，沒有本地性偏好。
• RACK_LOCAL 數據和程式碼處於同一個機架上的不同機器。這時，數據和程式碼處於不同機器上，需要通過網路傳遞，但還是在同一個機架上，一般也就通過一個交換機傳輸即可。
• ANY 數據在網路中未知，即數據和程式碼不在同一個機架上。

Spark傾向於讓所有任務都具有最佳的數據本地性，但這並非總是可行的。某些情況下，可能會出現一些空閑的執行器（executor）沒有待處理的數據，那麼Spark可能就會犧牲一些數據本地性。有兩種可能的選項：a）等待已經有任務的CPU，待其釋放後立即在同一台機器上啟動一個任務；b）立即在其他節點上啟動新任務，並把所需要的數據複製過去。

通常，Spark會等待一會，看看是否有CPU會被釋放出來。一旦等待超時，則立即在其他節點上啟動並將所需的數據複製過去。數據本地性各個級別之間的回落超時可以單獨配置，也可以在統一參數內一起設定；詳細請參考 configuration page 中的 spark.locality 相關參數。如果你的任務執行時間比較長並且數據本地性很差，你就應該試試調大這幾個參數，不過默認值一般都能適用於大多數場景了。