Spark SQL技術(shù)架構(gòu)優(yōu)化實踐

1、Spark SQL技術(shù)架構(gòu)優(yōu)化實踐目錄Spark SQL 架構(gòu)簡介Spark SQL 引擎性能優(yōu)化Spark Shuffle 穩(wěn)定性提升與性能優(yōu)化總結(jié)與展望Spark SQL 架構(gòu)簡介Resolved Logical PlanUnresolved Logical PlanOptimized Logical PlanPhysical PlanCost ModelSelectedDAG PhysicalPlanRDDsCatalystDataFrameCatalogRBOCBOAEParserAnalyzerOptimizerQuery PlannerDatasetSQLBucket Join 改

2、進(jìn)物化列物化視密其它優(yōu)化Spark SQL 引擎優(yōu)化Spark SQL 引擎優(yōu)化Bucket 改進(jìn)partition 0partition 1partition 2Shuffle SortShuffle SortSort Merge Join partition 0partition 0partition 1partition 1partition 2partition 2partition npartition nTable 1partition mpartition 0partition 1partition 2Table 2partition kJoin 雙方數(shù)據(jù)分布不一致，需先 Shu

3、ffle 后 JoinSpark SQL 引擎優(yōu)化Bucket 改進(jìn)bucket 1bucket 2bucket nbucket 0bucket 1bucket 2Table 2bucket nJoin 雙方數(shù)據(jù)分布一致，無需 Shuffle 可直接 JoinTable 1Joinbucket 0HiveTableScanHiveTableScanSort Merge JoinoutputPartitioning: HashPartitioning(id, n, HiveHash) outputOrdering: SortOrder(id)requireChildDistribution: H

4、ashClusteredDistribution(id, n, HiveHash) requireChildOrdering: SortOrder(id)Spark SQL 引擎優(yōu)化Bucket 改進(jìn)ExchangeExchangeSortSortSort Merge JoinrequireChildDistribution: HashClusteredDistribution(id, n, Murmur3Hash) requireChildOrdering: SortOrder(id)outputPartitioning: UnknownPartitioning outputOrdering

5、: NilHiveTableScanHiveTableScanSpark SQL 引擎優(yōu)化Bucket 改進(jìn)HiveMMMM.RRRbucket 0bucket 1bucket (n-1).Spak SQLMMMM.bucket 0bubcukcekte0t 0bbuuckcketet10bucket (n-1).改進(jìn)一：與 Hive 兼容Spark SQL 引擎優(yōu)化Bucket 改進(jìn)改進(jìn)二：支持倍數(shù)關(guān)系 Bucket JoinTable A (3 bucket)Table B(6 bucket)bucket 1bucket 5bucket 2bucket 3bucket 4(0, 6, 12

6、)(1, 7, 13)(2, 8, 14)(3, 9, 15)(4, 10, 16)(5, 11, 17)(0, 3, 6, 9, 12, 15)(2, 5, 8, 11, 14, 17)(1, 4, 7, 10,13, 16)TableScanSort Merge JoinSortTable A (3 bucket)Table B (6 bucket)bucket 0bucket 0bucket 1bucket 2TableScanSpark SQL 引擎優(yōu)化Bucket 改進(jìn)改進(jìn)二：支持倍數(shù) Bucket JoinTable A (3 bucket)Table B(6 bucket)(0,

7、 6, 12)(1, 7, 13)(2, 8, 14)(3, 9, 15)(4, 10, 16)(5, 11, 17)(0, 3, 6, 9, 12, 15)(2, 5, 8, 11, 14, 17(1, 4, 7, 10,13, 16)Sort Merge Joinbucket 0bucket 0bucket 1bucket 1)bucket 2bucket 2bucket 0bucket 3bucket 1bucket 4bucket 2bucket 5(0, 3, 6, 9, 12, 15)(2, 5, 8, 11, 14, 17)(1, 4, 7, 10,13, 16)TableSca

8、nTableScanTable A (3 bucket)Table B (6 bucket)TableScanTable A (3 bucket)BucketUnionSpark SQL 引擎優(yōu)化Bucket 改進(jìn)改進(jìn)三：支持 Bucket Join 降級由千數(shù)據(jù)體量漲幅大，Bucket 個數(shù)不變的清況下會發(fā)生 Bucket 文件持續(xù)增大的問題為了解決這個問題，需要增加 Bucket 個數(shù)Bucket 個數(shù)修改后，讀歷史數(shù)據(jù)時會出問題需要支持非 Bucket 轉(zhuǎn)為 Bucket 表解決方案修改 Bucket 個數(shù)時，在 Table Property 里記錄修改日期只有所查詢的所有分區(qū)都在該日期

9、之后，才使用 Bucket Join，否則降級為 SortMergeJoinSpark SQL 引擎優(yōu)化Bucket 改進(jìn)X23X11X42Y85Y67Y73Z28Z52Z43AX11X23X42Y67Y73Y85Z28Z43Z52CTableScanSort Merge Join on A BTable X Bucket by ATable Y Bucket by ATableScan改進(jìn)四：支持超集Table XBucket by ABCTable Y Bucket by AABSort on A BSort on A BSpark SQL 引擎優(yōu)化物化列Spark SQL 讀取復(fù)雜類型列

10、(Map/Struct/Array) 存在以 下問題讀取大量不必要的數(shù)據(jù)復(fù)雜類型不能進(jìn)行向量化讀取復(fù)雜類型列上的 Filter 無法下推重復(fù)計算：JSON 字符串字段提取CPU消耗大Spark SQL 引擎優(yōu)化物化列業(yè)務(wù)側(cè)解決方案基千主表維護(hù)一張新的表， 將高頻 key 作為獨立的列自衷大量數(shù)據(jù)表存在復(fù)雜類型字段：Map/ Array/ Struct Example基礎(chǔ)表：event_log (按 date/hour/app分區(qū)）實際查詢基千 event和 params (MapType) 的 key 進(jìn)行過濾專用子表：stg_app_event_log(按 event 分區(qū)）維護(hù)方角度：專用

11、子表維護(hù)成本高: 60TB memory+3000 core / hour，180TB disk /day無法更換高頻 key：己獨立15個key， 8個key未被獨立，更換困難用戶角度(分析師 運營等）對千用戶的使用不透明專用子表：31% (6628/21130) 直接查詢 params (可節(jié)省860h/天）基礎(chǔ)表：29% (21559/75203) 直接查詢 params(節(jié)省2800h/天）Spark SQL 引擎優(yōu)化物化列Spark SQL 引擎優(yōu)化物化列時間 / IO不使用物化列物化列 Rewrite速度讀取數(shù)據(jù)量SQL_adhoc_66.3 min / 797.6 GB3.4 m

12、in / 111.8 GB85.3%86% SQL_adhoc_716.5 min / 3.2 TB5.0 min / 111.1 GB230%96.6%SQL_etl_224 min / 3.7 TB9.1 min / 686.1 GB130.8%82%物化列實際收益提升查詢性能Spark SQL 引擎優(yōu)化物化視圖解決的問題OLAP 奎詢中，會經(jīng)常基于表的某些固定字段進(jìn)行 Aggregate， Join 等耗時操作，造成大 量重復(fù)性計算，浪費資涌，并會影響奎詢的響應(yīng)時間物化視圖預(yù)先計算并保存 Aggregate， Join 等耗時操作的結(jié)果按需更新，保證與涌表數(shù)據(jù)的一致性對用戶完全透明，由S

13、QL引擎完成：物化視圖匹配、最優(yōu)物化視圖選取、Query Rewrite支持部分匹配支持 RollupSpark SQL 引擎優(yōu)化物化視圖CREATE TABLE user_event ( user STRING,event_type STRING, num INT,date INT,hour INT) USING parquet PARTITIONED BY (date, hour)CREATE METERIALIZED VIEWdate_hour_user_agg PARTITIONED BY (date)ON TABLE user_eventAS SELECT date, user, a

14、vg(num) FROM user_eventGROUP BY date, userSELECT user, sum(num) FROM user_event WHERE date = 20190101GROUP BY userSELECT user, sum_num FROM date_hour_user_agg WHERE date = 20190101透明 rewriteSpark SQL 引擎優(yōu)化其它優(yōu)化自適應(yīng)系列自適應(yīng)選擇 Partial / Final Aggregator自適應(yīng)設(shè)置 Split 大小Adaptive Execution索引系列Bloom FilterRTOLoca

15、l Sort物化系列物化列物化視圖：預(yù)聚合物化視圖，Bucket 物化視圖運行時優(yōu)化系列Spark Shuffle 優(yōu)化嵌套類型優(yōu)化新壓縮格式支持 物理計劃優(yōu)化Runtime FilterShuffle 肖除 邏輯計劃優(yōu)化MultiJoin基于 HDFS 的 Shuffle 穩(wěn)定性提升Shuffle 性能優(yōu)化實踐與探索Spark Shuffle 穩(wěn)定性提升與性能優(yōu)化Spark Shuffle 穩(wěn)定性提升與性能優(yōu)化Shuffle Map StageShuffle Reduce StageShuffle Mapper 1Partition 0Partition 1Partition 2Shuffl

16、e Mapper 2Partition 0Partition 1Partition 2Shuffle Mapper 3Partition 0Partition 1Partition 2Partition 0Partition 1Partition 2Shuffle Reducer 1Shuffle Reducer 2Shuffle Reducer 3Shuffle原理Spark Shuffle 穩(wěn)定性提升與性能優(yōu)化問題一：Shuffle 失敗Spark Shuffle 穩(wěn)定性提升與性能優(yōu)化問題二：Shuffle Read 慢Spark Shuffle 穩(wěn)定性提升與性能優(yōu)化cpu.busy =

17、99%cpu.user = 97%Node Manager CPU 使用過高造成 Shuffle Read 超時Spark Shuffle 穩(wěn)定性提升與性能優(yōu)化Node Manager CPU 使用過高造成 Shuffle Read 超時Spark Shuffle 穩(wěn)定性提升與性能優(yōu)化DataNode 將磁盤 IO 打滿造成 Shuffle Read 失敗Spark Shuffle 穩(wěn)定性提升與性能優(yōu)化Shuffle 數(shù)據(jù)直接寫 HDFSSpark Shuffle 穩(wěn)定性提升與性能優(yōu)化Shuffle數(shù)據(jù)量: 800GBMapper:1200 Reducer:2200Shuffle 讀寫性能無差

18、異Shuffle數(shù)據(jù)量:450GBMapper:10000 Reducer:10000 Shuffle Read 過程變慢VSSpark Shuffle 穩(wěn)定性提升與性能優(yōu)化60萬45萬30萬10000 mapper x 10000 reducer x 2 = 2億 RPCSpark Shuffle 穩(wěn)定性提升與性能優(yōu)化Shuffle ServiceHDFSShuffle Mapper 1Local DiskShuffle WriterShuffle Mapper 2Local DiskShuffle WriterShuffle Mapper NLocal DiskShuffle WriterShuffle Reducer 1Shuffle ReaderShuffle Reducer 2Shuffle ReaderShuffle Reducer NShuffle ReaderShuffle 數(shù)據(jù)復(fù)制一份到 HDFS,Shuffle 失敗時從 HDFS 讀取Spark Shuffle 穩(wěn)定性提升與性能優(yōu)化57%+14%+18%+12%+基于 HDFS 的 Shuffle 穩(wěn)定性 提升方案已全量上線 近一年覆蓋的 Shuffle數(shù)據(jù)比例整體 性能提升夭級作業(yè) 性能提升小時級作業(yè)性 能提升Spark Shuffle 穩(wěn)定性提升與性能優(yōu)化基于 Mapper數(shù)據(jù)合并的優(yōu)化將多個 M