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

1、Spark SQL技術(shù)架構(gòu)優(yōu)化實(shí)踐目錄Spark SQL 架構(gòu)簡(jiǎn)介Spark SQL 引擎性能優(yōu)化Spark Shuffle 穩(wěn)定性提升與性能優(yōu)化總結(jié)與展望Spark SQL 架構(gòu)簡(jiǎn)介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ù)分布一致，無(wú)需 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 降級(jí)由千數(shù)據(jù)體量漲幅大，Bucket 個(gè)數(shù)不變的清況下會(huì)發(fā)生 Bucket 文件持續(xù)增大的問(wèn)題為了解決這個(gè)問(wèn)題，需要增加 Bucket 個(gè)數(shù)Bucket 個(gè)數(shù)修改后，讀歷史數(shù)據(jù)時(shí)會(huì)出問(wèn)題需要支持非 Bucket 轉(zhuǎn)為 Bucket 表解決方案修改 Bucket 個(gè)數(shù)時(shí)，在 Table Property 里記錄修改日期只有所查詢的所有分區(qū)都在該日期

9、之后，才使用 Bucket Join，否則降級(jí)為 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ù)雜類(lèi)型列

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

11、子表維護(hù)成本高: 60TB memory+3000 core / hour，180TB disk /day無(wú)法更換高頻 key：己獨(dú)立15個(gè)key， 8個(gè)key未被獨(dú)立，更換困難用戶角度(分析師 運(yùn)營(yíng)等）對(duì)千用戶的使用不透明專用子表：31% (6628/21130) 直接查詢 params (可節(jié)省860h/天）基礎(chǔ)表：29% (21559/75203) 直接查詢 params(節(jié)省2800h/天）Spark SQL 引擎優(yōu)化物化列Spark SQL 引擎優(yōu)化物化列時(shí)間 / 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%物化列實(shí)際收益提升查詢性能Spark SQL 引擎優(yōu)化物化視圖解決的問(wèn)題OLAP 奎詢中，會(huì)經(jīng)?；诒淼哪承┕潭ㄗ侄芜M(jìn)行 Aggregate， Join 等耗時(shí)操作，造成大 量重復(fù)性計(jì)算，浪費(fèi)資涌，并會(huì)影響奎詢的響應(yīng)時(shí)間物化視圖預(yù)先計(jì)算并保存 Aggregate， Join 等耗時(shí)操作的結(jié)果按需更新，保證與涌表數(shù)據(jù)的一致性對(duì)用戶完全透明，由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ùn)行時(shí)優(yōu)化系列Spark Shuffle 優(yōu)化嵌套類(lèi)型優(yōu)化新壓縮格式支持 物理計(jì)劃優(yōu)化Runtime FilterShuffle 肖除 邏輯計(jì)劃優(yōu)化MultiJoin基于 HDFS 的 Shuffle 穩(wěn)定性提升Shuffle 性能優(yōu)化實(shí)踐與探索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)化問(wèn)題一：Shuffle 失敗Spark Shuffle 穩(wěn)定性提升與性能優(yōu)化問(wèn)題二：Shuffle Read 慢Spark Shuffle 穩(wěn)定性提升與性能優(yōu)化cpu.busy =

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

18、異Shuffle數(shù)據(jù)量:450GBMapper:10000 Reducer:10000 Shuffle Read 過(guò)程變慢VSSpark Shuffle 穩(wěn)定性提升與性能優(yōu)化60萬(wàn)45萬(wàn)30萬(wàn)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 失敗時(shí)從 HDFS 讀取Spark Shuffle 穩(wěn)定性提升與性能優(yōu)化57%+14%+18%+12%+基于 HDFS 的 Shuffle 穩(wěn)定性 提升方案已全量上線 近一年覆蓋的 Shuffle數(shù)據(jù)比例整體 性能提升夭級(jí)作業(yè) 性能提升小時(shí)級(jí)作業(yè)性 能提升Spark Shuffle 穩(wěn)定性提升與性能優(yōu)化基于 Mapper數(shù)據(jù)合并的優(yōu)化將多個(gè) M