關于sparksql調優的一些操作

1、查看執行計劃
1、直接sql查看: explain select ... from ...
2、ds.explain()
2、執行計劃的處理流程
sql代碼 -> 未決斷的邏輯執行計劃 -> 根據元資料生成已決斷的邏輯執行計劃 -> 生成物理執行計劃 -> 模型評估 -> 選擇物理執行計劃 -> 生成執行代碼
3、cbo優化
注意:要想使用cbo優化必須先收集表、列的資訊,否則cbo優化不起作用
收集資訊:
收集表的資訊: ANALYZE TABLE 表名 COMPUTE STATISTICS
收集列的資訊: ANALYZE TABLE 表名 COMPUTE STATISTICS FOR COLUMNS 列1,列2,列3
使用cbo：
cbo優化可以優化多表join的順序、調整表的join的策略
spark.sql.cbo.enabled: 是否開啟cbo優化
spark.sql.cbo.joinReorder.enabled: 是否調整多表Join的順序
spark.sql.cbo.joinReorder.dp.threshold: 設定多表jion的表數量的閾值,一旦join的表數量超過該閾值則不優化多表join的順序
4、廣播join
場景: 大表join小表,可以將小表資料廣播出去避免shuffle操作
使用廣播join
1、設定小表的大小閾值:
在sparksession創建的時候使用config方法配置spark.sql.autoBroadcastJoinThreshold,后續join的時候一旦表的大小小于該閾值,則認為是小表,會廣播小表資料
2、使用hint語法強制廣播
select /*+ BROADCASTJOIN(別名1) */ .... from 表1 別名1 join 表2 別名2 on ...
select /*+ BROADCAST(別名1) */ .... from 表1 別名1 join 表2 別名2 on ...
select /*+ MAPJOIN(別名1) */ .... from 表1 別名1 join 表2 別名2 on ...
5、smb join
場景: 大表 join 大表可以使用smb join,可以減少join的時候掃描的表的資料量,提高效率
使用前提:
1、join的兩個大表必須是分桶表,兩個分桶表的桶的個數必須一樣或者是倍數關系
2、join的時候,join的列 = 分桶的的列
6、AQE自適應優化【spark3.x版本才有】
1、動態合并磁區: spark會根據磁區的資料量將小資料量的多個磁區合并成一個磁區,可以提高資源的利用率
spark.sql.adaptive.enabled: 是否開啟AQE優化
spark.sql.adaptive.coalescePartitions.enabled: 是否開啟動態合并磁區
spark.sql.adaptive.coalescePartitions.initialPartitionNum: 初始磁區數
spark.sql.adaptive.coalescePartitions.minPartitionNum: 合并之后的最小磁區數
當RDD的磁區數處于spark.sql.adaptive.coalescePartitions.initialPartitionNum與spark.sql.adaptive.coalescePartitions.minPartitionNum范圍內才會合并
spark.sql.adaptive.advisoryPartitionSizeInBytes: 合并磁區之后,磁區的資料量的預期大小
2、動態切換join策略: 在join的時候,會動態選擇性能最高的join策略,提高效率
spark.sql.adaptive.enabled: 是否開啟AQE優化
spark.sql.adaptive.localShuffleReader.enabled:在不需要進行shuffle重磁區時，嘗試使用本地shuffle讀取器，將sort-meger join 轉換為廣播join
3、動態申請資源: 當計算程序中資源不足會自動申請資源
spark.sql.adaptive.enabled: 是否開啟AQE優化
spark.dynamicAllocation.enabled: 是否開啟動態資源申請
spark.dynamicAllocation.shuffleTracking.enabled: 是否開啟shuffle狀態跟蹤
4、動態join資料傾斜: join的時候如果出現了資料傾斜,會動態調整磁區的資料量,優化資料傾斜導致的性能問題，
spark.sql.adaptive.enabled: 是否開啟AQE優化
spark.sql.adaptive.skewJoin.enable: 是否開啟join資料傾斜的優化
spark.sql.adaptive.skewJoin.skewedPartitionFactor: N
spark.sql.adaptive.skewJoin.skewedPartitionThresholdInBytes: M
spark.sql.adaptive.advisoryPartitionSizeInBytes: G [代表優化之后,磁區數資料的預期大小]
sparksql判斷出現資料傾斜的依據[需要兩個條件同時滿足]:
當某個磁區處理的資料量>= N * 所有task處理資料量的中位數
當某個磁區處理的資料量>= M
7、手動處理資料傾斜
1、如何判斷有資料傾斜
可以從4040webui界面查看所有task的執行時間與處理的資料量
如果某個task處理的資料量與執行時間相比其他task多很多,此時可以判定該task存在資料傾斜
2、資料傾斜的出現的階段
資料傾斜一般在shuffle階段才會出現
因為在shuffle的時候需要根據key.hashCode%磁區數得到磁區號,如果某個key的資料量很多,此時全部聚在一個磁區導致資料傾斜
3、資料傾斜出現的場景
1、key為null的資料很多
解決方案
1、如果null值資料對業務不影響,可以直接過濾掉
2、如果null值資料對業務影響,此時先可以給key弄一個亂數,將資料打散
2、groupBy的時候,某個key的資料很多導致資料傾斜
解決方案:
1、區域聚合
1、給分組的欄位加上亂數
2、按照加上亂數的key分組聚合
2、全域聚合
1、去掉key中的亂數
2、按照key再次磁區聚合
3、join的時候,某個key的資料很多導致資料傾斜
1、大表join小表出現資料傾斜
解決方案: 在sparksession中配置spark.sql.autoBroadcastJoinThreshold,將小表廣播出去,避免shuffle
2、大表 join 大表,個別key資料量多導致的資料傾斜
解決方案:
1、將傾斜表中傾斜key與非傾斜key資料單獨過濾出來
2、將非傾斜表中傾斜key與非傾斜key資料單獨過濾出來
3、將傾斜表中非傾斜key的資料與非傾斜表中非傾斜key的資料正常join
4、將傾斜表中傾斜key的資料的join的列加上N以內的亂數
5、將非傾斜表中傾斜key的資料擴容N倍
6、將第四步與第五步的資料join
7、將第六步的結果與第三步的結果union all
3、大表 join 大表,大量key資料量都比較多導致的資料傾斜
解決方案:
1、將傾斜表中join的列加上N[N最多只能是10以內]以內的亂數
2、將非傾斜表中的資料擴容N[N最多只能是10以內]倍
3、將第一步與第二步的資料join

標籤：其他

上一篇：Latex中也能展示動態圖？

下一篇：人臉識別、活體檢測（眨眼、搖頭、張嘴動作）