云計(jì)算大數(shù)據(jù)培訓(xùn)之Spark調(diào)優(yōu)（4）

1，盡量避免使用會(huì)發(fā)生shuffle的算子 
2，在數(shù)據(jù)源頭，對我們的數(shù)據(jù)提前進(jìn)行聚合

七，優(yōu)化數(shù)據(jù)結(jié)構(gòu) 
1、避免對象套對象 
2、減少數(shù)據(jù)集合的使用，盡量使用數(shù)組 
3、因?yàn)閷ο笮枰蛄谢?，能不用對象就盡量不用，建議使用字符串拼接 
4、可以使用第三方提供的占用內(nèi)存小，序列化速度快的數(shù)據(jù)結(jié)構(gòu)類庫，例如 fastUtil類庫

八，避免數(shù)據(jù)傾斜 
1，數(shù)據(jù)傾斜的概念 
有的時(shí)候，我們可能會(huì)遇到大數(shù)據(jù)計(jì)算中一個(gè)最棘手的問題——數(shù)據(jù)傾斜，此時(shí)spark作業(yè)性能會(huì)比期望查很多。數(shù)據(jù)傾斜調(diào)優(yōu)，就是使用各種技術(shù)方案解決不同類型的數(shù)據(jù)傾斜問題，一保證spark作業(yè)的性能

2，數(shù)據(jù)傾斜發(fā)生的現(xiàn)象： 
A、絕大多數(shù)task執(zhí)行的都非?？欤珎€(gè)別task執(zhí)行極慢。比如，總共有1000個(gè)task，997個(gè)task都在一分鐘內(nèi)執(zhí)行完了，但是剩余兩三個(gè)task卻要一兩個(gè)小時(shí)，這種情況很常見 
B、原本能夠正常執(zhí)行的spark作業(yè)，某天突然爆出 OOM（內(nèi)存溢出）異常，官場異常棧，是我們寫的業(yè)務(wù)代碼造成的，這種情況比較少見

3，數(shù)據(jù)傾斜發(fā)生的原理 
數(shù)據(jù)傾斜的原理很簡單:在進(jìn)行shuffle的時(shí)候，必須將各個(gè)節(jié)點(diǎn)上相同的key拉取到某個(gè)節(jié)點(diǎn)上的一個(gè)task來進(jìn)行處理，比如按照key進(jìn)行聚合或者join等操作，此時(shí)，如果某個(gè)key對應(yīng)的數(shù)據(jù)量特別大的話，就會(huì)發(fā)生數(shù)據(jù)傾斜，比如大部分key對應(yīng)10 條數(shù)據(jù)，但是個(gè)別的key卻對應(yīng)了100萬調(diào)數(shù)據(jù)，那么大部分task可能就只會(huì)分配到10-條數(shù)據(jù)，然后1秒就運(yùn)行完了，但是個(gè)別task可能分配到了100萬數(shù)據(jù)，要運(yùn)行一兩個(gè)小時(shí)。因此，整個(gè)spark作業(yè)的運(yùn)行進(jìn)度是由運(yùn)行時(shí)間最長的那個(gè)task決定的

4，然后定位導(dǎo)致數(shù)據(jù)傾斜的代碼 
數(shù)據(jù)傾斜只會(huì)發(fā)生在shuffle過程中，這里給大家羅列了一些常用的并且可能會(huì)觸發(fā)shuffle操作的算子：groupByKey, reduceByKey, aggregateByKey, join, cogroup, repartition等。出現(xiàn)數(shù)據(jù)傾斜時(shí)，可能就是你的代碼中使用了這些算子的某一個(gè)導(dǎo)致的

查看數(shù)據(jù)傾斜的數(shù)據(jù)：eventLogRDD.sample(false,0.1).countByKey().foreach(println(_))

5，數(shù)據(jù)傾斜的解決方案

解決方案一：使用HIve ETL預(yù)處理數(shù)據(jù)

方案使用場景：導(dǎo)致數(shù)據(jù)傾斜的是Hive表，如果該Hive表中的數(shù)據(jù)本身很不均勻（比如某個(gè)key對應(yīng)了100萬數(shù)據(jù)，其他的key才對應(yīng)了10條數(shù)據(jù)），而且業(yè)務(wù)場景需要頻繁使用spark對Hive表執(zhí)行某個(gè)分析操作，那么比較適合使用這種方案

方案實(shí)現(xiàn)思路：此時(shí)可以評估一下，是否可以通過Hive來進(jìn)行數(shù)據(jù)預(yù)處理（即通過Hive ETL預(yù)先對數(shù)據(jù)按照key進(jìn)行聚合，或者是預(yù)先和其他表進(jìn)行join），然后在spark作業(yè)中針對的數(shù)據(jù)源就不是原來的Hive表了，而是預(yù)處理后Hive表。此時(shí)由于數(shù)據(jù)已經(jīng)預(yù)先進(jìn)行了聚合或join操作，那么spark作業(yè)中也就不需要原先的shuffle類算子執(zhí)行的這類操作了

方案實(shí)現(xiàn)原理：這種方案從根本上解決了數(shù)據(jù)傾斜，因?yàn)閺氐妆苊饬嗽趕park中執(zhí)行shuffle類算子，那么就不會(huì)有數(shù)據(jù)傾斜的問題了，但是這里也要提醒一下大家，這種方式屬于治標(biāo)不治本，因?yàn)楫吘箶?shù)據(jù)本身就存在不均勻的問題，所以Hive ETL中進(jìn)行g(shù)roupBy 或者join等shuffle操作時(shí)，還是會(huì)出現(xiàn)數(shù)據(jù)傾斜，導(dǎo)致Hive ETL的速度很慢，我們只是把數(shù)據(jù)傾斜的發(fā)生提前到了Hive ETL中，避免了spark程序發(fā)生數(shù)據(jù)傾斜而已

方案優(yōu)點(diǎn)：實(shí)現(xiàn)起來簡單便捷，效果還非常好，完全規(guī)避了數(shù)據(jù)傾斜，spark作業(yè)的性能會(huì)大幅提升

方案缺點(diǎn)：治標(biāo)不治本，Hive ETL 中還是會(huì)發(fā)生數(shù)據(jù)傾斜

解決方案二：過濾少數(shù)導(dǎo)致傾斜的key

方案使用的場景：如果發(fā)現(xiàn)導(dǎo)致傾斜的key就少數(shù)幾個(gè)，而且對計(jì)算本身的影響并不是很大的話，那么很適合使用這種方案，比如99% 的key就對應(yīng)10條數(shù)據(jù)，但是只有一個(gè)key對應(yīng)了100萬數(shù)據(jù)，從而導(dǎo)致了數(shù)據(jù)傾斜

方案實(shí)現(xiàn)思路：如果我們判斷那少數(shù)的幾個(gè)數(shù)據(jù)量特別多的key，對作業(yè)的執(zhí)行和就算結(jié)果不是特別重要的話，那么干脆就直接過濾掉那少數(shù)的幾個(gè)key，比如，在spark SQL中可以使用where 子句過濾掉這些key或者在sparkRDD執(zhí)行filter算子過濾掉這些key，如果每次作業(yè)執(zhí)行時(shí)，動(dòng)態(tài)判定那些key的數(shù)據(jù)量最多然后再進(jìn)行過濾，那么可以直接使用sample算子對RDD進(jìn)行采樣，然后計(jì)算出每個(gè)key的數(shù)量，取數(shù)據(jù)量最多的key過濾掉即可

方案實(shí)現(xiàn)原理：將導(dǎo)致數(shù)據(jù)傾斜的key過濾掉之后，這些key就不參與計(jì)算了，自然不可能產(chǎn)生數(shù)據(jù)傾斜

方案優(yōu)點(diǎn)：實(shí)現(xiàn)簡單，而且效果也很好，可以完全規(guī)避掉數(shù)據(jù)傾斜。

方案缺點(diǎn)：使用場景不多，大多數(shù)情況下，導(dǎo)致傾斜的key還是很多的，并不是只有少數(shù)幾個(gè)

解決方案三：提高shuffle操作的并行度

方案使用場景：如果我們必須要對數(shù)據(jù)傾斜迎難而上，那么就建議優(yōu)先使用這種方案，因?yàn)檫@是處理數(shù)據(jù)傾斜最簡單的一種方案

方案實(shí)現(xiàn)思路：在對RDD執(zhí)行shuffle算子時(shí)，給shuffle算子傳入一個(gè)參數(shù)，比如：reduceByKey（+,1000），該參數(shù)就設(shè)置了這個(gè)shuffle算子執(zhí)行時(shí)shuffle read task的數(shù)量，對于sparkSQL中的shuffle類語句，比如GroupByKeyDemo，join，等，需要設(shè)置一個(gè)參數(shù)，即spark.sql.shuffle.partitions,該參數(shù)代表了shuffle read task 的并行度，該值默認(rèn)是200，對于很多場景來說都有點(diǎn)過小

方案實(shí)現(xiàn)原理：增加shuffle read task的數(shù)量，可以讓原本分配給一個(gè)task的多個(gè)key分配給多個(gè)task，從而讓每個(gè)task處理比原來更少的數(shù)據(jù)。舉例來說，如果有5個(gè)key，每個(gè)key對應(yīng)10條數(shù)據(jù)，這5個(gè)key都是分配給一個(gè)task的。那么這個(gè)task就要處理50條數(shù)據(jù)，而增加了shuffle read task 以后。每個(gè)task就分配到一個(gè)key，即每個(gè)task就處理10條數(shù)據(jù)，那么自然每個(gè)task的執(zhí)行時(shí)間都會(huì)變短了。

方案優(yōu)點(diǎn)：實(shí)現(xiàn)起來比較簡單，可以有效緩解和減輕數(shù)據(jù)傾斜的影響

方案缺點(diǎn)：只是緩解了數(shù)據(jù)傾斜而已，沒有徹底根除問題，根據(jù)時(shí)間來看，其效果有限

解決方案四：兩階段聚合（局部集合+全局聚合）

方案使用場景：對RDD執(zhí)行reduceByKey等聚合類shuffle算子或者在sparkSQL中使用GroupBy 語句進(jìn)行分組聚合時(shí)，比較適用這種場景

方案實(shí)現(xiàn)思路：這個(gè)方案的核心實(shí)現(xiàn)思路就是進(jìn)行兩階段聚合。第一次是局部聚合，先給每個(gè)key都打上一個(gè)隨機(jī)數(shù)，比如10以內(nèi)的隨機(jī)數(shù)，此時(shí)原先的key就變得不一樣了，比如（hello，1），（hello，1），（hello，1），（hello，1）… 就會(huì)變成（1_hello，1），（1_hello，1），（2_hello，1），（2_hello，1）… 接著對打上隨機(jī)數(shù)后的數(shù)據(jù)，執(zhí)行reduceByKey等聚合操作，進(jìn)行局部聚合，那么聚合結(jié)果，就會(huì)變成（1_hello，2），（2_hello，2）… 然后將各個(gè)key的前綴給去掉，就會(huì)變成（hello，2），（hello，2）… ，再次進(jìn)行全局聚合操作，就可以得到最終結(jié)果了，比如（hello，4）

方案實(shí)現(xiàn)原理：將原本相同的key通過附加隨機(jī)前綴的方式，變成多個(gè)不聽的key，就可以讓原本被一個(gè)task處理的數(shù)據(jù)分散到多個(gè)task上去做局部聚合，進(jìn)而解決單個(gè)task處理數(shù)據(jù)過多的問題，接著取出掉隨機(jī)前綴，再次進(jìn)行全局聚合，就可以得到最終的結(jié)果

方案優(yōu)點(diǎn)：對于聚合類的shuffle操作導(dǎo)致的數(shù)據(jù)傾斜，效果是非常不錯(cuò)的，通常都可以解決掉數(shù)據(jù)傾斜，或者至少是大幅度緩解數(shù)據(jù)傾斜，將spark作業(yè)的性能提升數(shù)倍以上

方案缺點(diǎn)：僅僅適用于聚合類的shuffle操作，適用范圍相對較窄。如果是join類的shuffle操作，還得用其他的解決方案

九，開啟推測機(jī)制

推測機(jī)制后，如果集群中，某一臺(tái)機(jī)器的幾個(gè)task特別慢，推測機(jī)制會(huì)將任務(wù)分配到其他機(jī)器執(zhí)行，最后spark會(huì)選取最快的作為最終結(jié)果

在spark-default.conf 中添加：spark.speculation true

推測機(jī)制與一下幾個(gè)參數(shù)有關(guān)： 
1、spark.speculation.interval 100: 檢測周期，單位毫秒 
2、Spark.speculation.quantile 0.75: 完成task的百分比時(shí)啟動(dòng)推測 
3、Spark.speculation.multiplier 1.5: 比其他的慢多少倍時(shí)啟動(dòng)推測

十，算子使用技巧

1、非必要的情況下，不要使用collect 
2、盡量使用，例如：foreachPartition 就不要使用foreach，在創(chuàng)建連接對象（數(shù)據(jù)連接對象等），盡量在分區(qū)上創(chuàng)建