1: Spark 整體架構(gòu)

Spark 是新一代的大數(shù)據(jù)處理引擎，支持批處理和流處理，也還支持各種機(jī)器學(xué)習(xí)和圖計算，它就是一個Master-worker 架構(gòu)，所以整個的架構(gòu)就如下所示：

2: Spark 任務(wù)提交命令
一般我們使用shell 命令提交，命令如下：

./bin/spark-submit \--master yarn  \--deploy-mode cluster \--driver-cores 2   \--driver-memory 1g \--executor-cores 4 \--num-executors 10 \--executor-memory 8g \--class PackageName.ClassName XXXX.jar \--name "Spark Job Name" \

driver 和executor 就是對應(yīng)的spark的 driver 和executor 的配置，然后再指定個部署模式master 就可以了。

3: spark提交任務(wù)的流程
如上所示，我們在提交spark 任務(wù)的時候，有個參數(shù)可供選擇模式，就是deploy-mode 的參數(shù)，它的值有client 和cluster 之分；client 說白了就是driver 在客戶端，任務(wù)的分配和調(diào)度都在客戶端，這種只適合用于測試；畢竟一旦流量大了的話客戶端是頂不住的啊；而cluster 模式是spark對應(yīng)的driver 和executor 都是在yarn集群上，相對來說穩(wěn)定，我們這里只著重說yarn-cluster 的提交流程

3.1: 首先如上面的第二點(diǎn)所示，我們先提交啟動命令
3.2: 然后客戶端這邊先運(yùn)行提交的jar包里面的main方法
3.3: 接著和RM 通信，告訴RM 啟動ApplicationMaster，./bin/ApplicatioMaster
3.4: RM 隨機(jī)選擇一臺NM 準(zhǔn)備啟動AM
3.5: 在AM 里面啟動driver，然后讓driver 進(jìn)行SparkContext 的初始化以及進(jìn)行任務(wù)的切分
3.6：AM 再向RM 申請資源
3.7: 申請資源后AM 就啟動ExecutionBackend,
3.8: ExecutionBackend 啟動 Executors
3.9：Driver 把任務(wù)分配給Executors
cluster 流程圖如下所示：

client 流程圖如下所示：

4: Spark 中的RDD
4.1: RDD 概念
還是老樣子，先理解一個東西之前我們看看它的學(xué)名是什么，Resilient Distribute DataSet ，就是彈性分布式數(shù)據(jù)集；說白了，它的本質(zhì)還是數(shù)據(jù)的集合，MR 和Flink的輸入是一條條數(shù)據(jù)，這Spark 說我乖點(diǎn)了，輸入我就先把它們整成一個個數(shù)據(jù)集；哈哈哈，符合預(yù)先聚合的思想。說白了，RDD 就是三個層次的抽象，Dataset, partition, 以及record；對應(yīng)到生活中的例子，就是班級，組以及單個同學(xué)；

4.2：RDD 的特點(diǎn)是啥？從學(xué)名可以窺見一二，彈性，分布式
我們先說彈性，說白了就是靈活，可以容錯，它是怎么容錯的呢？就是RDD之間互相有依賴關(guān)系，如果某個RDD的分區(qū)沒有了，可以從原始數(shù)據(jù)和依賴關(guān)系得來，不用讓其他分區(qū)重算；
那么分布式呢，就是一個RDD的分區(qū)數(shù)據(jù)可以分發(fā)到不同的節(jié)點(diǎn)上進(jìn)行計算；靠近數(shù)據(jù)的計算嘛；
一個RDD的分區(qū)數(shù)可以自定義；如果沒有自定義，則和hdfs的block數(shù)一樣，或者和cpu的核數(shù)成比例關(guān)系，一核心大概可以處理2-4個分區(qū)。

4.3: RDD 之間的血緣關(guān)系
血緣血緣，套到數(shù)據(jù)集這邊就知道是上下游RDD的關(guān)系了，這里的血緣指的就是子RDD和父RDD 之間是分區(qū)之間一對一還是多對一；說白了如果一個父RDD的一個分區(qū)的數(shù)據(jù)給到了子RDD的多個分區(qū)，這明擺著寬廣嘛，就是寬依賴；否則就是摘依賴，一對一的關(guān)系；如下所示

4.4: RDD 之間的stage 劃分
這里就有個問題了，為什么要進(jìn)行stage的劃分呢? 大家可以想想，這個spark的RDD 也是基于內(nèi)存的計算，如果一個任務(wù)一直這樣計算下去，比如算到90%的時候，機(jī)器突然宕機(jī)了，任務(wù)全部失敗了，是不是又得重新計算，所以我們就要進(jìn)行個中間數(shù)據(jù)的備份嘛，比如算完了一個stage的數(shù)據(jù)，先把它存儲到外部系統(tǒng)，這樣就不怕任務(wù)突然中斷了；

所以大家是不是結(jié)合上面的寬窄依賴就想到了，如果出現(xiàn)了寬依賴我們就截斷，劃分一個stage；畢竟可以先保存一份數(shù)據(jù)嘛，可能一個子RDD有多個父RDD，每個父RDD就自己先把數(shù)據(jù)保存起來，做個備份嘛；

所以我們stage 劃分就是根據(jù)寬窄依賴去劃分的，碰到寬依賴就劃分一個stage，stage 里面的算子和數(shù)據(jù)自成一個天地。

4.5: spark的編程思想之RDD
所以兄弟們可以看到，說白了，我們進(jìn)行spark的編程的時候，就是基于RDD的，然后和算子一起，把RDD當(dāng)作點(diǎn)，這些算子當(dāng)成邊，不就構(gòu)建成了我們的這個有向無環(huán)圖(DAG) 嘛；沒有相互依賴的RDD 進(jìn)行一個并行計算，有相互依賴的窄依賴的類型也可以進(jìn)行并行計算，當(dāng)碰到寬依賴的時候，就要進(jìn)行數(shù)據(jù)打散了，不就是shuffle嘛，最后算好的數(shù)據(jù)落盤就行了。

5: Spark中g(shù)roupByKey，reduceByKey，, combineByKey, aggregrateByKey的區(qū)別

怎么說呢，其實(shí)這些算子都是針對數(shù)據(jù)做聚合的操作，groupByKey 和reduceByKey 沒有定義初始值的結(jié)構(gòu)，groupByKey 默認(rèn)用的hash 分區(qū)，而reduceByKey 可以自定義分區(qū)，也會提前進(jìn)行combine；這兩個算子分區(qū)內(nèi)的邏輯和分區(qū)間的邏輯都是一致的；
而aggregrateByKey 和 combineByKey 和 則在reduceByKey的基礎(chǔ)上做了些變化，aggregrateByKey 是自定義分區(qū)內(nèi)和分區(qū)間的邏輯；而combineByKey 是在上述的基礎(chǔ)上也增加一個初始值自定義。

6: spark中的hashShuffle,sortShuffle, 以及優(yōu)化后的這兩個變種

6.1: 普通的hashShuffle ，說白了就是在shuffle的過程中使用hash對key 進(jìn)行分組，假如上游有100個類似的mapTask，下游有100個reduce task，每個mapTask 會針對下游生成100個文件，總共就是10000個小文件了，這個對于集群來說負(fù)擔(dān)就太大了，如下所示：

6.2: 優(yōu)化后的HashShuffle
上述的普通版產(chǎn)生的小文件個數(shù)太多了，所以我們需要優(yōu)化下，優(yōu)化的重點(diǎn)思路就是復(fù)用，既然每個上游的task都要生成這么多文件，可不可以一個Executor 里面的緩沖區(qū)復(fù)用呢？答案是可以的，spark 官方就根據(jù)cpu core 的數(shù)量 * 下一個stage的task個數(shù)來確定緩沖區(qū)數(shù)量以及文件個數(shù)，相對來說少了挺多的，原先要10000個，現(xiàn)在可能只要5*100個就可以了。哈哈哈，不過這個優(yōu)化機(jī)制要通過spark.shuffle.consolidateFiles=true 開啟；

6.3: sortShuffle
sortShuffle 是什么呢？說白了就是在shuffle 之前先進(jìn)行排序，然后也會有溢寫到磁盤，可能會生成多個文件，但是最終每個task會對所有的文件進(jìn)行合并，最終只生成一個文件和文件對應(yīng)的索引，讓下游的task 根據(jù)索引文件去找數(shù)據(jù)，拉取數(shù)據(jù)。如下所示：

這個對應(yīng)的文件更少了，對下游的壓力也更小了。

6.4: byPassSortShuffle
這個和上面的sortShuffle 如出一轍，唯一的區(qū)別就是在寫文件的時候不會進(jìn)行排序，省去了這部分的開銷；不過這個需要觸發(fā)的點(diǎn)有兩個，
6.4.1: shuffle map task 數(shù)量小于spark.shuffle.sort.bypassMergeThreshold=200參數(shù)的值。
6.4.2: 不是聚合類的 shuffle 算子

兄弟們最終是不是還想問問，現(xiàn)階段該怎么啟用sortShuffle，spark2.x 已經(jīng)把所有的shuffle都默認(rèn)成了sortShuffle了。

7: spark中的cache,persist 和checkpoint 的區(qū)別
三者都是把數(shù)據(jù)持久化的，前面兩個是把數(shù)據(jù)緩存到內(nèi)存，而checkpoint 是吧數(shù)據(jù)存到hdfs 或者本地文件；前者不會截斷血緣關(guān)系，后者會截斷，畢竟hdfs副本可以容錯；cache 實(shí)際上就是調(diào)用的persist

8: RDD , DataFrame, 以及DataSet 的區(qū)別

8.1: RDD 就是我們剛剛說的彈性分布式數(shù)據(jù)集嘛，是面向?qū)ο蟮?#xff0c;是需要序列化和反序列化的，在網(wǎng)絡(luò)IO中性能消耗大，是不支持Spark Sql的；但是它是編譯類型安全的；

8.2: DataFrame 怎么說呢？就是相當(dāng)于一個表，類似于python 和R，它只知道表頭，只知道下面的Name,age,Height；不知道具體每個字段對應(yīng)的數(shù)據(jù)類型，它就相當(dāng)于表里的v如下所示：

所以它是編譯時類型不安全的；不過它有個優(yōu)點(diǎn)，它的存儲是放在java的堆外內(nèi)存，不用進(jìn)行GC了，也不用進(jìn)行序列化和反序列化帶來的開銷了

8.3: 這就是天生為sql 創(chuàng)建的數(shù)據(jù)集了，它就相當(dāng)于sql里面的表，既知道表的字段頭，也知道每個表的字段的數(shù)據(jù)類型；而且它還支持序列化的時候輕量的序列化，它結(jié)合了DF 和 RDD的優(yōu)點(diǎn)，編譯時也是類型安全的，并且?guī)砹艘粋€全新的概念，Encode，它可以保證按需序列化數(shù)據(jù)了。

9: SparkShuffle 的優(yōu)化：
上一步我們講了spark上游MapTask生成小文件的優(yōu)化，那么接下來我們也可以看看sparkReduce的這個Reduce 端的優(yōu)化。
9.1: 既然要優(yōu)化reducer端，首先能不能讓每次buffer存儲的數(shù)據(jù)量多點(diǎn)，就節(jié)省了reduce 拉取的個數(shù)，這個值通過spark.shuffle.file.buffer 來控制，一般32k，可以是64k；
9.2: 既然要拉取的數(shù)據(jù)量少點(diǎn)，是不是可以提高每批次拉取的數(shù)據(jù)量呢？那就是spark.reducer.maxSizeInFlight 的值控制的，一般是48M，可以到96M；

10. sparkStreaming 和kafka的連接方式
說白了，這個問題就是問的sparkStreaming 如何從kafka 拉取數(shù)據(jù)的，
10.1：Receiver 方式，這個是Spark 1.5 之前的版本，說白了就是Executor 里面有個receiver的組件，它負(fù)責(zé)和kafka創(chuàng)建連接，并且使用kafka的高階api 去和kafka 通信，從kafka 消費(fèi)過來的數(shù)據(jù)基本上是先放在內(nèi)存中，然后供spark的executor 去做其他處理，所以數(shù)據(jù)量激增的時候有可能導(dǎo)致內(nèi)存炸掉； 畢竟數(shù)據(jù)放在內(nèi)存中很有可能丟失，是吧？所以它做了一個checkPoint，會對每個批次的數(shù)據(jù)做個備份，學(xué)名叫做WAL（Write ahead log） 機(jī)制，就是會把數(shù)據(jù)備份到分布式文件系統(tǒng)比如hdfs上面；
10.2：Direct 方式，直接每次輪詢kafka的partition的數(shù)據(jù)，有幾個partition 這邊也就輪詢幾次，使用kafka的低階的api，自己保存每個partition的offset，自己維護(hù)，保證了可以和kafka的同步；可以做到數(shù)據(jù)精確消費(fèi)一次；
生產(chǎn)環(huán)境一般都用DIrect 方式，畢竟不用checkpoint，不用降低數(shù)據(jù)處理速率，而且可以精準(zhǔn)消費(fèi)一次；不存在重復(fù)消費(fèi)的可能。
說白了，就是Receiver 是被動的，可能會被kafka的數(shù)據(jù)給干爆；Direct 是主動的，自己掌握offset ,去主動消費(fèi)

11：sparkStreaming 窗口函數(shù)的原理
因?yàn)閟parkStreaming 本質(zhì)上是微批數(shù)據(jù)當(dāng)作實(shí)時數(shù)據(jù)（如下所示）

val ssc = new StreamingContext(sc,Seconds(5))//5表示5秒中對數(shù)據(jù)進(jìn)行切分形成一個RDD
``
所以窗口數(shù)據(jù)也是微批的合集，窗口一定要是微批的整數(shù)倍，這樣才能算是可以統(tǒng)計出來的有效窗口

12: spark 常見算子解釋

12.1: mapToPair, 說白了就是把原來的數(shù)據(jù)轉(zhuǎn)換成關(guān)系對的形式，不管之前的數(shù)據(jù)怎么樣，轉(zhuǎn)換之后的數(shù)據(jù)就是pair對的形式，比如 tuple<String,Long>
12.2: transform, tranformToPair, 都是基于sparkStreaming 的流式數(shù)據(jù)結(jié)構(gòu)上的算子，transform可能就是返回一個流式的數(shù)據(jù)Dstream, 里面的數(shù)據(jù)單位是Object; 而tranformToPair 返回的是帶有pair 對的流是數(shù)據(jù)。
12.3: spark 編程，最關(guān)鍵的是你要知道你的輸入是什么，你要輸出什么，是tuple呢，還是String呢，還是其他數(shù)據(jù)呢，RDD里面裝載的基礎(chǔ)數(shù)據(jù)結(jié)構(gòu)要知道；

12.3:
自定義排序類就是要實(shí)現(xiàn)Ordered 以及 Serialziable 接口，定義好大于，小于，compreTo的邏輯；
自定義的accumulator 說白了就是一個累加器，繼承實(shí)現(xiàn)AccumulatorParam的方法，最關(guān)鍵的是要知道這個累加器的邏輯是什么，對哪個值累加，在addAccumulator 中實(shí)現(xiàn)這個累加的邏輯；
自定義的聚合函數(shù)，也就是udaf的function , 是要繼承UserDefinedAggregateFunction的類，最重要的是要做三件事，初始化傳進(jìn)來的值，在update方法里面對每個分區(qū)的值做業(yè)務(wù)判斷處理;
在merge 方法里面對所有分區(qū)的值做個融合判斷處理，最終在buffer中更新

13:spark電商項(xiàng)目的核心

13.1: 首先從業(yè)務(wù)方面可以分為session訪問指標(biāo)計算（top10 session），頁面轉(zhuǎn)換率計算（點(diǎn)擊/下單/支付）；廣告相關(guān)的指標(biāo)計算(top10 廣告，省份廣告)，訂單指標(biāo)(省份銷量高的產(chǎn)品，屬性銷量高的產(chǎn)品) 的相關(guān)統(tǒng)計計算；產(chǎn)品相關(guān)(各區(qū)域內(nèi)產(chǎn)品點(diǎn)擊次數(shù)) 的指標(biāo)計算；核心點(diǎn)是和產(chǎn)品經(jīng)理核對好各個指標(biāo)的計算方法；
13.2: 大數(shù)據(jù)大數(shù)據(jù)，最重要的關(guān)注點(diǎn)就是我們要關(guān)注性能，因?yàn)楹玫男阅芎筒畹男阅芘苋蝿?wù)的時間可以相差到幾個甚至十幾個小時，所以我們在編寫代碼的時候就要盡可能考慮到這些點(diǎn)；從系統(tǒng)可用資源方面，代碼層面，數(shù)據(jù)傾斜和組成方面進(jìn)行考慮，盡量達(dá)到最大的性能；盡量考慮穩(wěn)定性。

13.3: spark調(diào)優(yōu)
13.3.1: 算子調(diào)優(yōu)
13.3.2: shuffle 調(diào)優(yōu)
13.3.3: 資源調(diào)優(yōu)
13.3.4: jvm 調(diào)優(yōu)，降低cache 內(nèi)存的比例

14: Spark 內(nèi)存管理機(jī)制
14.1: 靜態(tài)內(nèi)存變?yōu)閯討B(tài)內(nèi)存，統(tǒng)一內(nèi)存管理了，分成了Storage Memory 和Execution Memory ，分別是1:1 的關(guān)系，存儲數(shù)據(jù)，存儲shuffle過程中的數(shù)據(jù)