一. MaxCompute Spark 介紹

MaxCompute Spark是MaxCompute提供的相容開源的Spark計算服務。它在統一的計算資源和資料集權限體系之上，提供Spark計算架構，支援使用者以熟悉的開發使用方式送出運作Spark作業，以滿足更豐富的資料處理分析場景。

1.1 關鍵特性

支援原生多版本Spark作業

- 社群原生Spark運作在MaxCompute裡，完全相容Spark的API，支援多個Spark版本同時運作

統一的計算資源

- 像MaxCompute SQL/MR等任務類型一樣，運作在MaxCompute項目開通的統一計算資源中

統一的資料和權限管理

- 遵循MaxCompute項目的權限體系，在通路使用者權限範圍内安全地查詢資料

與開源系統相同的使用體驗

- 提供原生的開源實時Spark UI和查詢曆史日志的功能

1.2 系統結構

原生Spark通過MaxCompute Cupid平台能夠在MaxCompute中運作

MaxCompute Spark 使用和常見問題一. MaxCompute Spark 介紹二. 開發環境搭建三. 配置介紹四. 作業診斷五. 常見問題六. 相關文檔

1.3 限制與限制

目前MaxCompute Spark支援以下适用場景：

- 離線計算場景：GraphX、Mllib、RDD、Spark-SQL、PySpark等
- Streaming場景
- 讀寫MaxCompute Table
- 引用MaxCompute中的檔案資源
- 讀寫VPC環境下的服務，如RDS、Redis、HBase、ECS上部署的服務等
- 讀寫OSS非結構化存儲

使用限制

- 不支援互動式類需求Spark-Shell、Spark-SQL-Shell、PySpark-Shell等
- 不支援通路MaxCompute外部表，函數和UDF
- 隻支援Local模式和Yarn-cluster模式運作

二. 開發環境搭建

2.1 運作模式

通過Spark用戶端送出

- Yarn-Cluster模式，送出任務到MaxCompute叢集中
- Local模式

通過Dataworks送出

- 本質上也是Yarn-Cluster模式，送出任務到MaxCompute叢集中

2.2 通過用戶端送出

2.2.1 Yarn-Cluster模式

下載下傳MC Spark用戶端

- Spark 1.6.3
- Spark 2.3.0

環境變量配置

## JAVA_HOME配置
# 推薦使用JDK 1.8
export JAVA_HOME=/path/to/jdk
export CLASSPATH=.:$JAVA_HOME/lib/dt.jar:$JAVA_HOME/lib/tools.jar
export PATH=$JAVA_HOME/bin:$PATH

## SPARK_HOME設定
# 下載下傳上文提到的MaxCompute Spark用戶端并解壓到本地任意路徑
# 請不要直接設定SPARK_HOME等于以下路徑下述路徑僅做展示用途
# 請指向正确的路徑
export SPARK_HOME=/path/to/spark_extracted_package
export PATH=$SPARK_HOME/bin:$PATH

## PySpark配置Python版本
export PATH=/path/to/python/bin/:$PATH

參數配置

- 将$SPARK_HOME/conf/spark-defaults.conf.template 重命名為 spark-defaults.conf
- 參數配置參考下文

準備項目工程

git clone https://github.com/aliyun/MaxCompute-Spark.git
cd spark-2.x
mvn clean package

任務送出

// bash環境
cd $SPARK_HOME
bin/spark-submit --master yarn-cluster --class com.aliyun.odps.spark.examples.SparkPi \
/path/to/MaxCompute-Spark/spark-2.x/target/spark-examples_2.11-1.0.0-SNAPSHOT-shaded.jar

// 在windows環境送出的指令
cd $SPARK_HOME/bin
spark-submit.cmd --master yarn-cluster --class com.aliyun.odps.spark.examples.SparkPi
\path\to\MaxCompute-Spark\spark-2.x\target\spark-examples_2.11-1.0.0-SNAPSHOT-shaded.jar

2.2.2 Local模式

與Yarn Cluster模式類似，使用者首先需要做以上準備工作

## Java/Scala
cd $SPARK_HOME
./bin/spark-submit --master local[4] --class com.aliyun.odps.spark.examples.SparkPi \
/path/to/odps-spark-examples/spark-examples/target/spark-examples-2.0.0-SNAPSHOT-shaded.jar

## PySpark
cd $SPARK_HOME
./bin/spark-submit --master local[4] \
/path/to/odps-spark-examples/spark-examples/src/main/python/odps_table_rw.py

IDEA調試注意

- IDEA運作Local模式是不能直接引用spark-defaults.conf裡的配置，需要手動在代碼裡指定相關配置
- 一定要注意需要在IDEA裡手動添加MaxCompute Spark用戶端的相關依賴（jars目錄），否則會出現以下報錯：

the value of spark.sql.catalogimplementation should be one of hive in-memory but was odps

- 參考文檔

2.3 通過DataWorks送出

2.3.1 資源上傳

本質上MC Spark節點的配置對應于spark-submit指令的參數和選項

ODPS SPARK節點	spark-submit
主Java、Python資源	app jar or python file
配置項	--conf PROP=VALUE
Main Class	--class CLASS_NAME
參數	[app arguments]
選擇JAR資源	--jars JARS
選擇Python資源	--py-files PY_FILES
選擇File資源	--files FILES
選擇Archives資源	--archives

上傳資源：

- 0～50MB：可以直接在DataWorks界面建立資源并上傳
- 50MB～500MB：可以先利用MaxCompute用戶端(CMD)上傳，然後在DataWorks界面添加到資料開發，參考

資源引用：

- 資源送出後，可以在DataWorks Spark節點界面選擇需要的資源（jar/python/file/archive）
- 任務運作時：資源檔案預設會上傳到Driver和Executor的目前工作目錄

2.3.2 參數和配置

Spark 配置項：對應于spark-submit指令的--conf選項

- accessid，accesskey，projectname，endpoint，runtime.end.point，task.major.version無需配置
- 除此之外，需要将spark-default.conf中的配置逐條加到dataworks的配置項中

給主類傳參數(如bizdate)

- 首先在排程->參數中添加參數，然後在Spark節點“參數”欄引用該參數。多個參數用空格分隔
- 該參數會傳給使用者主類，使用者在代碼中解析該參數即可

三. 配置介紹

3.1 配置的位置

3.1.1 Spark配置的位置

使用者使用Maxcompute Spark通常會有幾個位置可以添加Spark配置，主要包括：

- 位置1：spark-defaults.conf，使用者通過用戶端送出時在spark-defaults.conf檔案中添加的Spark配置
- 位置2：dataworks的配置項，使用者通過dataworks送出時在配置項添加的Spark配置，這部配置設定置最終會在位置3中被添加
- 位置3：配置在啟動腳本spark-submit --conf選項中
- 位置4：配置在使用者代碼中，使用者在初始化SparkContext時設定的Spark配置

Spark配置的優先級

- 使用者代碼 > spark-submit --選項 > spark-defaults.conf配置 > spark-env.sh配置 > 預設值

3.1.2 需要區分的兩種配置

一種是必須要配置在spark-defaults.conf或者dataworks的配置項中才能生效（在任務送出之前需要），而不能配置在使用者代碼中，這類配置主要的特征是：

- 與Maxcompute/Cupid平台相關：一般參數名中都會帶odps或者cupid，通常這些參數與任務送出/資源申請都關系：

- - 顯而易見，一些資源擷取（如driver的記憶體，core，diskdriver，maxcompute資源），在任務執行之前就會用到，如果這些參數設定在代碼中，很明顯平台沒有辦法讀到，是以這些參數一定不要配置在代碼中
  - 其中一些參數即使配置在代碼中，也不會造成任務失敗，但是不會生效
  - 其中一些參數配置在代碼中，可能會造成副作用：如在yarn-cluster模式下設定spark.master為local

- 通路VPC的參數：

- - 這類參數也與平台相關，打通網絡是在送出任務時就進行的

一種是在以上三種位置配置都可以生效，但是在代碼配置的優先級最高
推薦把任務運作與優化相關的參數配置在代碼中，而與資源平台相關的配置都配置在spark-defaults.conf或者dataworks的配置項中。

3.2 資源相關的參數

spark.executor.instances	總共申請的executor數目普通任務十幾個或者幾十個足夠了，若是處理大量資料時可以申請多一些，100—2000+
spark.executor.cores	每個executor的核數 Job的最大并行度是executor數目*executor core數
spark.executor.memory	代表申請executor的記憶體
spark.yarn.executor.memoryOverhead	申請executor的堆外記憶體，預設機關是MB 主要用于JVM自身，字元串, NIO Buffer等開銷單個executor的總記憶體是：spark.executor.memory+spark.yarn.executor.memoryOverhead
spark.driver.cores	類似executor
spark.driver.memory
spark.yarn.driver.memoryOverhead
spark.driver.maxResultSize	預設1g，控制worker送回driver的資料大小，一旦超過該限制，driver會終止執行
spark.hadoop.odps.cupid.disk.driver.device_size	代表本地網盤大小，預設值為20g 當出現No space left on device時可适當調大該值，最大支援100g 設定該參數需要包含機關‘g’

3.3 平台相關的參數

spark.hadoop.odps.project.name	Spark任務運作所在的project
spark.hadoop.odps.access.id	送出spark任務的accessId
spark.hadoop.odps.access.key	送出spark任務的accessKey
spark.hadoop.odps.end.point	用于任務送出中國公共雲一般設定為 http://service.cn.maxcompute.aliyun.com/api
spark.hadoop.odps.runtime.end.point	用于任務運作 http://service.cn.maxcompute.aliyun-inc.com/api
spark.hadoop.odps.task.major.version	代表目前使用的平台版本公共雲設定為cupid_v2即可
spark.sql.catalogImplementation	Spark 2.3版本需要設定為odps Spark 2.4以後版本會改為hive 為了便于作業遷移，建議不要将該配置寫在代碼中
spark.hadoop.odps.cupid.resources	該配置項指定了程式運作所需要的 Maxcompute資源，格式為<projectname>.<resourcename>，可指定多個，用逗号分隔。指定的資源将被下載下傳到driver和executor的工作目錄，經常使用該參數來引用較大的檔案。資源下載下傳到目錄後預設的名字是<projectname>.<resourcename> 如果需要重新命名，需要在配置時通過<projectname>.<resourcename>:<new resource name>進行重命名
spark.hadoop.odps.cupid.vectorization.enable	是否開啟向量化讀寫，預設為true
spark.hadoop.odps.input.split.size	用于調節讀Maxcompute表的并發度預設每個分區為256MB，該參數機關為MB
spark.hadoop.odps.cupid.vpc.domain.list	vpc通路依賴的參數，傳統的通路vpc的方式
spark.hadoop.odps.cupid.smartnat.enable	vpc通路依賴的參數如果region是北京或者上海，将該參數設定為true
spark.hadoop.odps.cupid.eni.enable	如果使用者已開通專線，則需要配置為true
spark.hadoop.odps.cupid.eni.info	如果使用者已開通專線，則需要設定該參數該參數代表使用者打通的vpc
spark.hadoop.odps.cupid.engine.running.type	普通作業3天沒跑完就會被強制回收，流式作業需要設定此值為longtime
spark.hadoop.odps.cupid.job.capability.duration.hours	流式作業權限檔案expired時間，機關小時
spark.hadoop.odps.moye.trackurl.dutation	流式作業jobview expired時間，機關小時

四. 作業診斷

4.1 Logview

4.1.1 Logview 介紹

在任務送出時會列印日志: 日志中含有logview連結 (關鍵字 logview url)
Master以及Worker的StdErr列印的是spark引擎輸出的日志，StdOut中列印使用者作業輸出到控制台的内容

4.1.2 利用Logview 排查問題

拿到Logview，一般首先看Driver的報錯，Driver會包含一些關鍵性的錯誤
如果Driver中出現類或者方法找不到的問題，一般是jar包打包的問題
如果Driver中出現連接配接外部VPC或者OSS出現Time out，這種情況一般要去排查一下參數配置
如果Driver中出現連接配接不到Executor，或者找不到Chunk等錯誤，通常是Executor已經提前退出，需要進一步檢視Executor的報錯，可能存在OOM

- 根據End Time做排序，結束時間越早，越容易是發生問題的Executor節點
- 根據Latency做排序，Latency代表了Executor的存活的時間，存活時間越短的，越有可能是根因所在

4.2 Spark UI和HistoryServer

Spark UI與社群版一緻，在logivew的summary子產品下找到Spark UI連結：

Spark UI的使用與社群原生版是一緻的，可以參考
注意

- Spark UI需要鑒權，隻有送出任務的Owner才能打開
- Spark UI僅在作業運作時才能打開，如果任務已經結束，那麼Spark UI是無法打開的，這時候需要檢視Spark History Server UI

五. 常見問題

1. local模式運作的問題

問題一：the value of spark.sql.catalogimplementation should be one of hive in-memory but was odps

- 原因在于使用者沒有正确地按照文檔将Maxcompute Spark的jars目錄添加到類路徑，導緻加載了社群版的spark包，需要按照文檔将jars目錄添加到類路徑

問題二：IDEA Local模式是不能直接引用spark-defaults.conf裡的配置，必須要把Spark配置項寫在代碼中
問題三：通路OSS和VPC：

- Local模式是處于使用者本機環境，網絡沒有隔離。而Yarn-Cluster模式是處于Maxcompute的網絡隔離環境中，必須要要配置vpc通路的相關參數
- Local模式下通路oss的endpoint通常是外網endpoint，而Yarn-cluster模式下通路vpc的endpoint是經典網絡endpoint

2. jar包打包的問題

java/scala程式經常會遇到Java類找不到/類沖突問題：

- 類沖突：使用者Jar包與Spark或平台依賴的Jar包沖突
- 類沒有找到：使用者Jar包沒有打成Fat Jar或者由于類沖突引起

打包需要注意：

- 依賴為provided和compile的差別：

- - provided：代碼依賴該jar包，但是隻在編譯的時候需要用，而運作時不需要，運作時會去叢集中去尋找的相應的jar包
  - compile：代碼依賴該jar包，在編譯、運作時候都需要，在叢集中不存在這些jar包，需要使用者打到自己的jar包中。這種類型的jar包一般是一些三方庫，且與spark運作無關，與使用者代碼邏輯有關

- 使用者送出的jar包必須是Fat jar：

- - 必須要把compile類型的依賴都打到使用者jar包中，保證代碼運作時能加載到這些依賴的類

需要設定為provided的jar包

- groupId為org.apache.spark的Jar包
- 平台相關的Jar包

- - cupid-sdk
  - hadoop-yarn-client
  - odps-sdk

需要設定為compile的jar包

- oss相關的jar包

- - hadoop-fs-oss

- 使用者通路其他服務用到的jar包：

- - 如mysql，hbase

- 使用者代碼需要引用的第三方庫

3. 需要引入Python包

很多時候使用者需要用到外部Python依賴

- 首先推薦使用者使用我們打包的公共資源，包含了常用的一些資料處理，計算，以及連接配接外部服務（mysql，redis，hbase）的三方庫

## 公共資源python2.7.13
spark.hadoop.odps.cupid.resources = public.python-2.7.13-ucs4.tar.gz
spark.pyspark.python = ./public.python-2.7.13-ucs4.tar.gz/python-2.7.13-ucs4/bin/python

## 公共資源python3.7.9
spark.hadoop.odps.cupid.resources = public.python-3.7.9-ucs4.tar.gz
spark.pyspark.python = ./public.python-3.7.9-ucs4.tar.gz/python-3.7.9-ucs4/bin/python3

- 如果不能滿足使用者需要，使用者可以在該公共資源的基礎上上傳wheel包
- 如果wheel包依賴鍊較為複雜，可以通過Docker容器進行打包

使用Docker容器打包：

- 為了保證與線上環境一緻，避免運作時so包找不到的問題，需要使用Docker容器進行打包
- Docker容器本質隻是提供了相容性較好的os環境，使用者需要在容器中進行打包，并将整個Python目錄壓縮後上傳到MaxCompute Resource中，最後在Spark任務中直接引用即可
- 參見

4. 需要引入外部檔案

需要引用到外部檔案的場景

- 使用者作業需要讀取一些配置檔案
- 使用者作業需要額外的jar包/Python庫

可以通過兩種方式上傳資源：

- 通過Spark參數上傳檔案
- 通過MaxCompute Resource上傳檔案

- MaxCompute Spark支援Spark社群版原生的--jars，--py-files等參數，可以在作業送出時通過這些參數将檔案上傳，這些檔案在任務運作時會被上傳到使用者的工作目錄下
- 通過DataWorks添加任務需要的資源，參見上文

MaxCompute Resource

- spark.hadoop.odps.cupid.resources參數，可以直接引用MaxCompute中的資源，這些資源在任務運作時也會被上傳到使用者的工作目錄下
- 使用方式

（1）通過MaxCompute用戶端将檔案上傳(單個檔案最大支援500MB)

（2）在Spark作業配置中添加spark.hadoop.odps.cupid.resources參數：格式為<projectname>.<resourcename>，如果需要引用多個檔案，需要用逗号隔開

（3）如果需要重命名，格式為<projectname>.<resourcename>:<new resource name>

如何讀取上傳的檔案：

- 如果需要讀取上傳的檔案資源，檔案路徑如下：

val dir = new File(".")
val targetFile = "file://" + dir.getCanonicalPath + "/" +檔案名

- 或者直接通過類加載器擷取檔案路徑，然後再讀取

5. VPC通路的問題

Maxcompute Spark是獨立運作在Maxcompute叢集的，網絡與外界隔離，是以無法直接通路vpc和公網，需要添加以下配置。
北京和上海Region使用smartnat

- 需要配置

- - spark.hadoop.odps.cupid.smartnat.enable=true

- 通路公網：假如要通路google.com:443，需要做以下兩步：

- - 提工單設定 project 級别白名單，把 google.com:443 加到odps.security.outbound.internetlist
  - 配置作業級别的公網通路白名單:spark.hadoop.odps.cupid.internet.access.list=google.com:443

其他Region：

- 隻需要配置spark.hadoop.odps.cupid.vpc.domain.list
- 無法通路公網

注意事項：

- vpc.domain.list 需要壓縮成一行，不能包含空格
- 支援同時通路同一個Region下的多個VPC，需要配置所有要通路的ip:port的白名單
- 需要在要通路的服務中添加ip白名單，允許100.104.0.0/16網段的通路
- 使用者要保證所有可能通路到的IP都已經加到vpc.domain.list，例如如果使用者要通路位于hdfs，hbase這種多個節點的服務，一定要把所有的節點都添加進來，不然可能會遇到Time out

6. OOM的問題

可能出現OOM的情況：

- 錯誤1: 在某些Executor中出現Cannot allocate memory，一般是系統記憶體不足，此時可以調整spark.yarn.executor.memoryOverhead參數，注意該參數是會計算到總記憶體數的，也不需要一次性增加太多，小心調整即可
- 錯誤2：Executor抛出java.lang.OutOfMemoryError: Java heap space
- 錯誤3：GC overhead limit exceeded
- 錯誤4：No route to host: workerd*********/Could not find CoarseGrainedScheduler，這類錯誤一般是一些Executor提前退出。如果一個task處理的資料非常大，容易發生OOM

Driver OOM：Driver OOM的可能性比較小，但是也是有可能出現的

- 如果需要使用collect算子将RDD的資料全部拉取到Driver上進行處理，那麼必須確定Driver的記憶體足夠大，否則會出現OOM記憶體溢出的問題。
- SparkContext，DAGScheduler都是運作在Driver端的。Stage切分也是在Driver端運作，如果使用者程式有過多的步驟，切分出過多的Stage，這部分資訊消耗的是Driver的記憶體，這個時候就需要調大Driver的記憶體。有時候如果stage過多，Driver端可能會有棧溢出的問題

一些解決方法：

- 限制executor 并行度，将cores 調小：多個同時運作的 Task 會共享一個Executor 的記憶體，使得單個 Task 可使用的記憶體減少，調小并行度能緩解記憶體壓力
- 增加單個Executor記憶體
- 增加分區數量，減少每個executor負載
- 考慮資料傾斜問題，因為資料傾斜導緻某個 task 記憶體不足，其它 task 記憶體足夠

7. No space left on device

這個錯誤意味這本地磁盤不足，通常這個報錯會在executor上出現，并導緻executor挂掉
解決方案

- 直接增加更多的磁盤空間：預設driver和executor都各提供20g的本地磁盤，當磁盤空間不足時可以調整spark.hadoop.odps.cupid.disk.driver.device_size
- 如果調整本地磁盤大小到100g後，仍然報該錯誤，說明單個executor寫的shuffle資料已經超過上限，可能是遇到了資料傾斜，這種情況下可以對資料重分區。或者增加executor的數量

8. 申請資源的問題

申請不到資源的幾種現象：

（1）在driver端一般會打以下日志

- WARN YarnClusterScheduler: Initial job has not accepted any resources; check your cluster UI to ensure that workers are registered and have sufficient resources

（2）在logview中隻能看到driver，而worker數量為0

（3）在spark ui中隻能看到driver，而worker數量為0

解決方案：

- 調整任務資源：調整使用者申請的executor總數或者單個executor的資源數量（一般是記憶體），如果單個executor請求的記憶體過多可能不太容易申請到
- 合理安排任務執行時間

其他注意事項：

- 必須配置spark.master=yarn-cluster才會正确的申請資源

9. 其他問題

如何切換Spark版本

- 版本号規則介紹：示例 spark-2.3.0-odps0.32.5

- - spark-2.3.0 是社群版本的spark版本号，Maxcompute Spark基于該社群版本進行适配
  - odps0.32.5 是Maxcompute Spark的小版本号，随着小版本号的更新，可能進行一些bug修複和sdk的更新

- 使用者送出作業的的Spark版本可能有以下幾種情況：

- - 情況1：直接通過本地用戶端送出任務，spark版本就是使用者本地用戶端的版本
  - 情況2：使用者通過dataworks送出任務，取決于dataworks gateway的預設spark版本，目前公共雲dataworks 公共資源組gateway的預設版本是 spark-2.3.0-odps0.32. 1
  - 情況3：使用者通過dataworks送出任務，配置參數spark.hadoop.odps.spark.version，則會按照配置的版本号來尋找對應的spark用戶端，使用者可以配置spark.hadoop.odps.spark.version= 5手動切換版本
  - 情況4：該情況優先級最高，使用者可以在本地用戶端或者是dataworks送出任務時配置以下參數，則類加載的優先級最高，是以會在spark任務啟動時優先使用該版本的spark

spark.hadoop.odps.cupid.resources = public.__spark_libs__2.3.0odps0.32.5.zip spark.driver.extraClassPath = ./public.__spark_libs__2.3.0odps0.32.5.zip/* spark.executor.extraClassPath = ./public.__spark_libs__2.3.0odps0.32.5.zip/*

需要在代碼中通路配置項：

- spark開頭的參數直接通過SparkConf類提供的接口直接讀取即可

Spark History Server渲染速度慢

- 可以添加壓縮配置：spark.eventLog.compress=true

如何正确地Kill一個運作中的Spark任務

- 通常通過兩種方式kill正在運作的Spark任務

（1）通過odps cmd 執行 kill + instanceId;

（2）通過dataworks界面執行stop

- 注意，直接在spark用戶端或者dataworks的任務送出界面執行Ctrl + C是無法kill一個Spark任務的

日志中文亂碼，添加以下配置

- spark.executor.extraJavaOptions=-Dfile.encoding=UTF-8 -Dsun.jnu.encoding=UTF-8
- spark.driver.extraJavaOptions=-Dfile.encoding=UTF-8 -Dsun.jnu.encoding=UTF-8
- 如果是pyspark作業需要設定下如下兩個參數：

- - spark.yarn.appMasterEnv.PYTHONIOENCODING=utf8
  - spark.executorEnv.PYTHONIOENCODING=utf8
  - 另外在python腳本的最前面加上如下的代碼：

# -*- coding: utf-8 -*-
import sys
reload(sys)
sys.setdefaultencoding('utf-8')

六. 相關文檔

MC Spark github wiki： https://github.com/aliyun/MaxCompute-Spark/wiki
Spark UI： https://spark.apache.org/docs/latest/web-ui.html
Spark 配置： https://spark.apache.org/docs/2.4.5/configuration.html

MaxCompute Spark 使用和常見問題一. MaxCompute Spark 介紹二. 開發環境搭建三. 配置介紹四. 作業診斷五. 常見問題六. 相關文檔