一、前言

Spark 作為大數(shù)據(jù)計(jì)算引擎，憑借其快速、穩(wěn)定、簡(jiǎn)易等特點(diǎn)，快速的占領(lǐng)了大數(shù)據(jù)計(jì)算的領(lǐng)域。本文主要為作者在搭建使用計(jì)算平臺(tái)的過程中，對(duì)于 Spark 的理解，希望能給讀者一些學(xué)習(xí)的思路。文章內(nèi)容為介紹 Spark 在 DataMagic 平臺(tái)扮演的角色、如何快速掌握 Spark 以及 DataMagic 平臺(tái)是如何使用好 Spark 的。

二、Spark 在 DataMagic 平臺(tái)中的角色

整套架構(gòu)的主要功能為日志接入、查詢 (實(shí)時(shí)和離線)、計(jì)算。離線計(jì)算平臺(tái)主要負(fù)責(zé)計(jì)算這一部分，系統(tǒng)的存儲(chǔ)用的是 COS(公司內(nèi)部存儲(chǔ))，而非 HDFS。

[[231286]]

圖 2-1

下面將主要介紹 Spark on Yarn 這一架構(gòu)，抽取出來即圖 2-2 所示，可以看到 Spark on yarn 的運(yùn)行流程。

圖 2-2

三、如何快速掌握 Spark

對(duì)于理解 Spark，我覺得掌握下面 4 個(gè)步驟就可以了。

1. 理解 Spark 術(shù)語

對(duì)于入門，學(xué)習(xí) Spark 可以通過其架構(gòu)圖，快速了解其關(guān)鍵術(shù)語，掌握了關(guān)鍵術(shù)語，對(duì) Spark 基本上就有認(rèn)識(shí)了，分別是結(jié)構(gòu)術(shù)語 Shuffle、Patitions、MapReduce、Driver、Application Master、Container、Resource Manager、Node Manager 等。API 編程術(shù)語關(guān)鍵 RDD、DataFrame，結(jié)構(gòu)術(shù)語用于了解其運(yùn)行原理，API 術(shù)語用于使用過程中編寫代碼，掌握了這些術(shù)語以及背后的知識(shí)，你就也知道 Spark 的運(yùn)行原理和如何編程了。

2. 掌握關(guān)鍵配置

Spark 在運(yùn)行的時(shí)候，很多運(yùn)行信息是通過配置文件讀取的，一般在 spark-defaults.conf，要把 Spark 使用好，需要掌握一些關(guān)鍵配置，例如跟運(yùn)行內(nèi)存相關(guān)的，spark.yarn.executor.memoryOverhead、spark.executor.memory，跟超時(shí)相關(guān)的 spark.network.timeout 等等，Spark 很多信息都可以通過配置進(jìn)行更改，因此對(duì)于配置需要有一定的掌握。但是使用配置時(shí)，也要根據(jù)不同的場(chǎng)景，這個(gè)舉個(gè)例子，例如 spark.speculation 配置，這個(gè)配置主要目的是推測(cè)執(zhí)行，當(dāng) worker1 執(zhí)行慢的情況下，Spark 會(huì)啟動(dòng)一個(gè) worker2，跟 worker1 執(zhí)行相同的任務(wù)，誰先執(zhí)行完就用誰的結(jié)果，從而加快計(jì)算速度，這個(gè)特性在一般計(jì)算任務(wù)來說是非常好的，但是如果是執(zhí)行一個(gè)出庫到 Mysql 的任務(wù)時(shí)，同時(shí)有兩個(gè)一樣的 worker，則會(huì)導(dǎo)致 Mysql 的數(shù)據(jù)重復(fù)。因此我們?cè)谑褂门渲脮r(shí)，一定要理解清楚，直接 google spark conf 就會(huì)列出很多配置了。

3. 使用好 Spark 的并行

我們之所以使用 Spark 進(jìn)行計(jì)算，原因就是因?yàn)樗?jì)算快，但是它快的原因很大在于它的并行度，掌握 Spark 是如何提供并行服務(wù)的，從而是我們更好的提高并行度。

對(duì)于提高并行度，對(duì)于 RDD，需要從幾個(gè)方面入手，1、配置 num-executor。2、配置 executor-cores。3、配置 spark.default.parallelism。三者之間的關(guān)系一般為 spark.default.parallelism=num-executors*executor-cores 的 2~3 倍較為合適。對(duì)于 Spark-sql，則設(shè)置 spark.sql.shuffle.partitions、num-executor 和 executor-cores。

4. 學(xué)會(huì)如何修改 Spark 代碼

新手而言，特別是需要對(duì) Spark 進(jìn)行優(yōu)化或者修改時(shí)，感到很迷茫，其實(shí)我們可以首先聚焦于局部，而 Spark 確實(shí)也是模塊化的，不需要覺得 Spark 復(fù)雜并且難以理解，我將從修改 Spark 代碼的某一角度來進(jìn)行分析。

首先，Spark 的目錄結(jié)構(gòu)如圖 3-1 所示，可以通過文件夾，快速知道 sql、graphx 等代碼所在位置，而 Spark 的運(yùn)行環(huán)境主要由 jar 包支撐，如圖 3-2 所示，這里截取部分 jar 包，實(shí)際上遠(yuǎn)比這多，所有的 jar 包都可以通過 Spark 的源代碼進(jìn)行編譯，當(dāng)需要修改某個(gè)功能時(shí)，僅需要找到相應(yīng) jar 包的代碼，修改之后，編譯該 jar 包，然后進(jìn)行替換就行了。

圖 3-1

圖 3-2

而對(duì)于編譯源代碼這塊，其實(shí)也非常簡(jiǎn)單，安裝好 maven、scala 等相關(guān)依賴，下載源代碼進(jìn)行編譯即可，掌握修改源碼技巧對(duì)于使用好開源項(xiàng)目十分重要。

四、DataMagic 平臺(tái)中的 Spark

Spark 在 DataMagic 中使用，也是在邊使用邊探索的過程，在這過程中，列舉了其比較重要的特點(diǎn)。

1. 快速部署

在計(jì)算中，計(jì)算任務(wù)的數(shù)量以及數(shù)據(jù)的量級(jí)每天都會(huì)發(fā)生變化，因此對(duì)于 Spark 平臺(tái)，需要有快速部署的特性，在實(shí)體機(jī)上，有一鍵部署腳本，只要運(yùn)行一個(gè)腳本，則可以馬上上線一個(gè)擁有 128G 內(nèi)存、48cores 的實(shí)體機(jī)，但是實(shí)體機(jī)通常需要申請(qǐng)報(bào)備才能獲得，因此還會(huì)有 docker 來支持計(jì)算資源。

2. 巧用配置優(yōu)化計(jì)算

Spark 大多數(shù)屬性都是通過配置來實(shí)現(xiàn)的，因此可以通過配置動(dòng)態(tài)修改 Spark 的運(yùn)行行為，這里舉個(gè)例子，例如通過配置自動(dòng)調(diào)整 exector 的數(shù)量。

2.1 在 nodeManager 的 yarn-site.xml 添加配置

yarn.nodemanager.aux-services

mapreduce_shuffle,spark_shuffle

yarn.nodemanager.aux-services.spark_shuffle.class

org.apache.spark.network.yarn.YarnShuffleService

2.2 將 spark-2.2.0-yarn-shuffle.jar 文件拷貝到 hadoop-yarn/lib 目錄下 (即 yarn 的庫目錄)

2.3 在 Spark 的 spark-default.xml 添加配置

spark.dynamicAllocation.minExecutors 1 #最小 Executor 數(shù)

spark.dynamicAllocation.maxExecutors 100 #*** Executor 數(shù)

通過這種配置，可以達(dá)到自動(dòng)調(diào)整 exector 的目的。

3. 合理分配資源

作為一個(gè)平臺(tái)，其計(jì)算任務(wù)肯定不是固定的，有的數(shù)據(jù)量多，有的數(shù)據(jù)量少，因此需要合理分配資源，例如有些千萬、億級(jí)別的數(shù)據(jù)，分配 20 核計(jì)算資源就足夠了。但是有些數(shù)據(jù)量級(jí)達(dá)到百億的，就需要分配更多的計(jì)算資源了。參考第三章節(jié)的第 3 點(diǎn)。

4. 貼合業(yè)務(wù)需求

計(jì)算的目的其實(shí)就是為了服務(wù)業(yè)務(wù)，業(yè)務(wù)的需求也理應(yīng)是平臺(tái)的追求，當(dāng)業(yè)務(wù)產(chǎn)生合理需求時(shí)，平臺(tái)方也應(yīng)該盡量去滿足。如為了支持業(yè)務(wù)高并發(fā)、高實(shí)時(shí)性查詢的需求下，Spark 在數(shù)據(jù)出庫方式上，支持了 Cmongo 的出庫方式。

sc = SparkContext(conf=conf) sqlContext = SQLContext(sc) database = d = dict((l.split('=') for l in dbparameter.split())) parquetFile = sqlContext.read.parquet(file_name) parquetFile.registerTempTable(tempTable) result = sqlContext.sql(sparksql) url = "mongodb://"+database['user']+":"+database['password']+"@"+database['host']+":"+database['port'] result.write.format("com.mongodb.spark.sql").mode('overwrite').options(uri=url,database=database['dbname'],collection=pg_table_name).save()

5. 適用場(chǎng)景

Spark 作為通用的計(jì)算平臺(tái)，在普通的應(yīng)用的場(chǎng)景下，一般而言是不需要額外修改的，但是 DataMagic 平臺(tái)上，我們需要 “在前行中改變”。這里舉個(gè)簡(jiǎn)單的場(chǎng)景，在日志分析中，日志的量級(jí)達(dá)到千億 / 日的級(jí)別，當(dāng)?shù)讓尤罩镜哪承┳侄纬霈F(xiàn) utf-8 編碼都解析不了的時(shí)候，在 Spark 任務(wù)中進(jìn)行計(jì)算會(huì)發(fā)生異常，然后失敗，然而如果在數(shù)據(jù)落地之前對(duì)亂碼數(shù)據(jù)進(jìn)行過濾，則有可能會(huì)影響數(shù)據(jù)采集的效率，因此最終決定在 Spark 計(jì)算過程中解決中這個(gè)問題，因此在 Spark 計(jì)算時(shí)，對(duì)數(shù)據(jù)進(jìn)行轉(zhuǎn)換的代碼處加上異常判斷來解決該問題。

6.Job 問題定位

Spark 在計(jì)算任務(wù)失敗時(shí)候，需要去定位失敗原因，當(dāng) Job 失敗是，可以通過 yarn logs -applicationId application 來合并任務(wù) log，打開 log，定位到 Traceback，一般可以找到失敗原因。一般而言，失敗可以分成幾類。

a. 代碼問題，寫的 Sql 有語法問題，或者 Spark 代碼有問題。

b. Spark 問題，舊 Spark 版本處理 NULL 值等。

c. 任務(wù)長(zhǎng)時(shí)間 Running 狀態(tài)，則可能是數(shù)據(jù)傾斜問題。

d. 任務(wù)內(nèi)存越界問題。

7. 集群管理

Spark 集群在日常使用中，也是需要運(yùn)營(yíng)維護(hù)的，從而運(yùn)營(yíng)維護(hù)，發(fā)現(xiàn)其存在的問題，不斷的對(duì)集群進(jìn)行優(yōu)化，這里從以下幾個(gè)方面進(jìn)行介紹，通過運(yùn)營(yíng)手段來保障集群的健壯性和穩(wěn)定性，保證任務(wù)順利執(zhí)行。

a. 定時(shí)查看是否有 lost node 和 unhealthy node，可以通過腳本來定時(shí)設(shè)置告警，若存在，則需要進(jìn)行定位處理。

b. 定時(shí)掃描 hdfs 的運(yùn)行 log 是否滿了，需要定時(shí)刪除過期 log。

c. 定時(shí)掃描集群資源是否滿足計(jì)算任務(wù)使用，能夠提前部署資源。

五、總結(jié)

本文主要是通過作者在搭建使用計(jì)算平臺(tái)的過程中，寫出對(duì)于 Spark 的理解，并且介紹了 Spark 在當(dāng)前的 DataMagic 是如何使用的，當(dāng)前平臺(tái)已經(jīng)用于架平離線分析，每天計(jì)算分析的數(shù)據(jù)量已經(jīng)達(dá)到千億~ 萬億級(jí)別。