業(yè)務(wù)背景
Spark讀取文件時(shí)，對(duì)于可切分的文件，會(huì)將文件切分為一系列 Split ，每個(gè)Split對(duì)應(yīng)一個(gè)Task。一般而言，Split的大小與HDFS的Block大小相當(dāng)，即128MB。
對(duì)于Parquet文件，因?yàn)槭前戳写鎯?chǔ)，在讀取數(shù)據(jù)時(shí)，可按列剪枝。而劃分Split時(shí)并未考慮列剪枝。因此當(dāng)所選取的列數(shù)遠(yuǎn)小于總列數(shù)時(shí)，實(shí)際每個(gè)Split內(nèi)需要讀取的數(shù)據(jù)量遠(yuǎn)小于Split大小，從而使得每個(gè)Task處理的數(shù)據(jù)量非常?。ǜ鶕?jù)實(shí)際測(cè)試，可能只有幾MB甚至幾KB）
因此可根據(jù)目標(biāo)列數(shù)占總列數(shù)的比例相應(yīng)增大Split大小，從而保證一個(gè)Split內(nèi)目標(biāo)列讀取數(shù)據(jù)總量與一個(gè)Block大小相當(dāng)，即128MB。
優(yōu)化方式
手動(dòng)設(shè)置參數(shù)
默認(rèn)配置
spark.sql.files.maxPartitionBytes=134217728 // 該參數(shù)決定Spark SQL讀取可切分文件時(shí)，每個(gè)Partition / Split的最大值，單位Byte。
spark.sql.files.openCostInBytes=4194304 // 該參數(shù)決定Spark SQL讀取文件時(shí)，每個(gè)Partition / Split的最小值，單位Byte。
通過(guò)如下SQL計(jì)算記錄總條數(shù)
select count(device_id) from log_daily where date='20210202' and hour='12'
總Task個(gè)數(shù)為8779個(gè)
Task讀取數(shù)據(jù)量中位數(shù)為2.7MB
通過(guò)如下參數(shù)設(shè)置，將Split大小擴(kuò)大10倍，每個(gè)Partition / Split的最小值擴(kuò)大4倍
set spark.sql.files.maxPartitionBytes= 1073741824;
set spark.sql.files.openCostInBytes= 16777216;
執(zhí)行后，總Task個(gè)數(shù)為1098個(gè)
Task讀取數(shù)據(jù)量中位數(shù)為擴(kuò)大為22.6MB
可以看到Task的并行度下降，使用資源量可以得到控制，同時(shí)執(zhí)行速度并沒(méi)有多大變化
自動(dòng)設(shè)置參數(shù)
通過(guò)如下參數(shù)設(shè)置，將開(kāi)啟自適應(yīng)文件切分
set spark.sql.parquet.adaptiveFileSplit=true;
例：一張表共10個(gè)字段，5個(gè)IntegerType（寬度為4），5個(gè)LongType（寬度為8），總寬度為60。
現(xiàn)對(duì)其中兩個(gè)字段進(jìn)行查詢，一個(gè)IntegerType，一個(gè)LongType，總寬度為12。故放大倍數(shù)為 60 / 12 = 5
注：該方案只對(duì)列式存儲(chǔ)有用。
