?本文從計(jì)算資源治理實(shí)踐出發(fā)，帶大家清楚認(rèn)識(shí)計(jì)算資源治理到底該如何進(jìn)行，并如何應(yīng)用到其他項(xiàng)目中。

01前言

由于數(shù)據(jù)治理層面可以分多個(gè)層面且內(nèi)容繁多（包括模型合規(guī)、數(shù)據(jù)質(zhì)量、數(shù)據(jù)安全、計(jì)算/存儲(chǔ)資源、數(shù)據(jù)價(jià)值等治理內(nèi)容），因此需要單獨(dú)拆分為6個(gè)模塊單獨(dú)去闡述其中內(nèi)容。

筆者作為數(shù)倉(cāng)開發(fā)經(jīng)常會(huì)收到大量集群資源滿載、任務(wù)產(chǎn)出延時(shí)等消息/郵件，甚至下游數(shù)分及其他同學(xué)也會(huì)詢問(wèn)任務(wù)運(yùn)行慢的情況，在這里很多數(shù)倉(cāng)同學(xué)遇到這類問(wèn)題第一想到的都是加資源解決，但事實(shí)真不一定是缺少資源，而是需要優(yōu)化當(dāng)前問(wèn)題任務(wù)。所以本期從團(tuán)隊(duì)做計(jì)算資源治理視角出發(fā)，帶大家清楚認(rèn)識(shí)計(jì)算資源治理到底該如何進(jìn)行。

02問(wèn)題出現(xiàn)

在做計(jì)算治理之前（2022.12）我們團(tuán)隊(duì)盤點(diǎn)了下當(dāng)前計(jì)算資源存在的幾個(gè)問(wèn)題：

（1）30+高消耗任務(wù)：由于數(shù)倉(cāng)前中期業(yè)務(wù)擴(kuò)張，要覆蓋大量場(chǎng)景應(yīng)用，存在大量問(wèn)題代碼運(yùn)行時(shí)數(shù)據(jù)傾斜，在消耗大量集群計(jì)算資源下，產(chǎn)出時(shí)間也久；

（2）200w+的小文件：當(dāng)前任務(wù)存在未合并小文件、任務(wù)Reduce數(shù)量過(guò)多、上游數(shù)據(jù)源接入（尤其是API數(shù)據(jù)接入）會(huì)造成過(guò)多小文件出現(xiàn)，小文件過(guò)多會(huì)開啟更多數(shù)據(jù)讀取，執(zhí)行會(huì)浪費(fèi)大量的資源，嚴(yán)重影響性能；

（3）任務(wù)調(diào)度安排不合理：多數(shù)任務(wù)集中在凌晨2-5點(diǎn)執(zhí)行且該區(qū)間CPU滿載，導(dǎo)致該時(shí)間段資源消耗成了重災(zāi)區(qū)，所有核心/非核心任務(wù)都在爭(zhēng)搶資源，部分核心任務(wù)不能按時(shí)產(chǎn)出一直在等待階段；

（4）線上無(wú)效DQC（數(shù)據(jù)質(zhì)量監(jiān)控）&監(jiān)控配置資源過(guò)?。捍嬖诓糠謿v史任務(wù)沒(méi)下線表及DQC場(chǎng)景，每日都在空跑無(wú)意義DQC浪費(fèi)資源，同時(shí)DQC資源過(guò)少導(dǎo)致DQC需要運(yùn)行過(guò)長(zhǎng)時(shí)間；

（5）重復(fù)開發(fā)任務(wù)/無(wú)用任務(wù)：早期協(xié)助下游做了較多煙囪數(shù)據(jù)模型，因?yàn)榉N種原因，部分任務(wù)不再被使用，煙囪模型分散加工導(dǎo)致資源復(fù)用率降低；

（6）任務(wù)缺少調(diào)優(yōu)參數(shù)&部分任務(wù)仍然使用MapReduce/Spark2計(jì)算引擎：任務(wù)缺少調(diào)優(yōu)參數(shù)導(dǎo)致資源不能適配及動(dòng)態(tài)調(diào)整，甚至線上仍有早期配置MapReduce/Spark2計(jì)算引擎導(dǎo)致運(yùn)行效率較低。

03思考與行動(dòng)

3.1 治理前的思考：?

在治理之前我想到一個(gè)問(wèn)題，切入點(diǎn)該從哪里開始最合適？

經(jīng)過(guò)與團(tuán)隊(duì)多次腦暴對(duì)當(dāng)前治理優(yōu)先級(jí)/改動(dòng)成本大小/難度做了一個(gè)排序，我們先選擇從簡(jiǎn)單的參數(shù)調(diào)優(yōu)&任務(wù)引擎切換開始->小文件治理->DQC治理->高消耗任務(wù)治理->調(diào)度安排->下線無(wú)用模型及沉淀指標(biāo)到其他數(shù)據(jù)資產(chǎn)，同時(shí)在初期我們完成各類元數(shù)據(jù)接入搭建治理看板以及團(tuán)隊(duì)治理產(chǎn)出統(tǒng)計(jì)數(shù)據(jù)模型，并通過(guò)網(wǎng)易數(shù)帆提供的數(shù)據(jù)治理平臺(tái)解決具體細(xì)節(jié)問(wèn)題。

（數(shù)據(jù)治理平臺(tái)截圖）

3.2 治理行動(dòng)：

（1）大部分任務(wù)切換至Spark3計(jì)算引擎&補(bǔ)充任務(wù)調(diào)優(yōu)參數(shù)

補(bǔ)充Spark調(diào)優(yōu)參數(shù)（參數(shù)內(nèi)容詳見文末），任務(wù)統(tǒng)一使用Spark3引擎加速，并充分利用Spark3的AQE特性及Z-Order排序算法特性。

AQE解釋：Spark 社區(qū)在 DAG Scheduler 中，新增了一個(gè) API 在支持提交單個(gè) Map 階段，以及在運(yùn)行時(shí)修改 shuffle 分區(qū)數(shù)等等，而這些就是 AQE，在 Spark 運(yùn)行時(shí)，每當(dāng)一個(gè) Shuffle、Map 階段進(jìn)行完畢，AQE 就會(huì)統(tǒng)計(jì)這個(gè)階段的信息，并且基于規(guī)則進(jìn)行動(dòng)態(tài)調(diào)整并修正還未執(zhí)行的任務(wù)邏輯計(jì)算與物理計(jì)劃（在條件運(yùn)行的情況下），使得 Spark 程序在接下來(lái)的運(yùn)行過(guò)程中得到優(yōu)化。

Z-Order解釋：Z-Order 是一種可以將多維數(shù)據(jù)壓縮到一維的技術(shù)，在時(shí)空索引以及圖像方面使用較廣，比如我們常用order by a,b,c 會(huì)面臨索引覆蓋的問(wèn)題，Z-Order by a,b,c 效果對(duì)每個(gè)字段是對(duì)等的

（2）小文件治理

在這里我們使用內(nèi)部數(shù)據(jù)治理平臺(tái)-數(shù)據(jù)治理360對(duì)存在小文件較多表提供內(nèi)容展示（本質(zhì)采集HDFS對(duì)應(yīng)路徑下文件數(shù)的日志去顯示）

當(dāng)前小文件處理：

對(duì)于分區(qū)較多使用Spark3進(jìn)行動(dòng)態(tài)分區(qū)刷新，（Spark3具備小文件自動(dòng)合并功能，如未使用Spark3可配置Spark3/Hive小文件合并參數(shù)刷新，參數(shù)詳見文末），代碼如下：

set hive.exec.dynamic.partition.mode=nonstrict;
insert overwrite table xxx.xxx partition (ds)
select column
,ds
from xxx.xxx

對(duì)于分區(qū)較少或未分區(qū)的表采用重建表，補(bǔ)數(shù)據(jù)方法回刷。

小文件預(yù)防：

• 使用Spark3引擎，自動(dòng)合并小文件
• 減少Reduce的數(shù)量(可以使用參數(shù)進(jìn)行控制)
• 用Distribute By Rand控制分區(qū)中數(shù)據(jù)量
• 添加合并小文件參數(shù) 
• 將數(shù)據(jù)源抽取后的表做一個(gè)任務(wù)（本質(zhì)也是回刷分區(qū)合并小文件任務(wù)）去處理小文件保障從數(shù)據(jù)源開始小文件不向下游流去

（3）DQC治理

無(wú)效DQC下線：難點(diǎn)在于需要查找所有DQC對(duì)應(yīng)的線上任務(wù)，查看該DQC任務(wù)是否與線上任務(wù)一一匹配，從而找到無(wú)效DQC任務(wù)下線，內(nèi)容繁雜耗時(shí)較多。

DQC資源：由于之前DQC配置資源為集群默認(rèn)參數(shù)，效率極低導(dǎo)致所有DQC運(yùn)行時(shí)長(zhǎng)均超過(guò)10min，從而使得整體任務(wù)鏈路運(yùn)行時(shí)長(zhǎng)過(guò)久，調(diào)整Driver內(nèi)存為2048M，Executor個(gè)數(shù)為2，Executor內(nèi)存為4096M

（4）高消耗任務(wù)調(diào)優(yōu)

這里存在2個(gè)難點(diǎn)：優(yōu)化效果不可控、高消耗任務(wù)調(diào)整到何種程度算合適，針對(duì)這個(gè)這個(gè)難點(diǎn)我們?nèi)∷泻诵臄?shù)據(jù)資產(chǎn)任務(wù)均值，保障單個(gè)任務(wù)消耗小于平均消耗，同時(shí)我們針對(duì)當(dāng)前高消耗任務(wù)列舉出如下可優(yōu)化的方式：

• 關(guān)聯(lián)表過(guò)多，需拆分
• 關(guān)聯(lián)時(shí)一對(duì)多，數(shù)據(jù)膨脹
• 資源配置過(guò)多，運(yùn)行時(shí)資源嚴(yán)重浪費(fèi)，需要將配置調(diào)?。ò―river內(nèi)存、Executor個(gè)數(shù)、Executor內(nèi)存）
• 代碼結(jié)尾添加Distribute By Rand（），用來(lái)控制Map輸出結(jié)果的分發(fā)
• 查詢中列和行未裁剪、分區(qū)未限定、Where條件未限定
• SQL中Distinct切換為Group by（Distinct會(huì)被hive翻譯成一個(gè)全局唯一Reduce任務(wù)來(lái)做去重操作，Group by則會(huì)被hive翻譯成分組聚合運(yùn)算，會(huì)有多個(gè)Reduce任務(wù)并行處理，每個(gè)Reduce對(duì)收到的一部分?jǐn)?shù)據(jù)組，進(jìn)行每組聚合（去重））
• 關(guān)聯(lián)后計(jì)算切換為子查詢計(jì)算好后再關(guān)聯(lián)
• 使用Map Join（Map Join會(huì)把小表全部讀入內(nèi)存中，在Map階段直接拿另外一個(gè)表的數(shù)據(jù)和內(nèi)存中表數(shù)據(jù)做匹配，由于在Map是進(jìn)行了Join操作，省去了Reduce運(yùn)行的效率也會(huì)高很多）可用參數(shù)代替

（5）任務(wù)調(diào)度合理優(yōu)化

對(duì)于調(diào)度優(yōu)化一開始會(huì)無(wú)從下手，統(tǒng)計(jì)凌晨2-5點(diǎn)區(qū)間下大概600+任務(wù)難梳理，同時(shí)存在任務(wù)依賴，修改起來(lái)可能會(huì)對(duì)下游整體有大的影響，因此我們選擇循序漸進(jìn)先梳理再改善。

• 找到所有表的輸出輸入點(diǎn)即啟始ODS與末尾ADS
• 劃分其中核心表/非核心表，及對(duì)應(yīng)任務(wù)開始時(shí)間與結(jié)束時(shí)間
• 按照梳理內(nèi)容把非核心的任務(wù)穿插在當(dāng)前集群資源非高峰時(shí)期（2點(diǎn)前與5點(diǎn)后），同時(shí)把核心任務(wù)調(diào)度提前，保障CDM層任務(wù)及時(shí)產(chǎn)出
• 對(duì)實(shí)踐后內(nèi)容再度調(diào)優(yōu)，達(dá)到資源最大利用率

（6）煙囪任務(wù)下沉&無(wú)用任務(wù)下線

煙囪表過(guò)多，需下沉指標(biāo)到DWS中提升復(fù)用性，對(duì)于無(wú)用任務(wù)也需要及時(shí)下線（這里需要拿到元數(shù)據(jù)血緣最好到報(bào)表層級(jí)的數(shù)據(jù)血緣，防止任務(wù)下線后導(dǎo)致可視化內(nèi)容問(wèn)題產(chǎn)生），減少開發(fā)資源消耗。

04治理效果

（1）Hive與Spark2任務(wù)升級(jí)Spark3.1，總計(jì)升級(jí)任務(wù)137個(gè),升級(jí)任務(wù)后總體任務(wù)執(zhí)行效率提升43%，cpu資源消耗降低41%，內(nèi)存資源消耗降低46%

（2）治理小文件數(shù)大于10000+以上的數(shù)倉(cāng)表總計(jì)30+張，小文件總數(shù)由216w下降至67w

（3）下線無(wú)效DQC任務(wù)總計(jì)50+，修改DQC配置資源降低運(yùn)行時(shí)長(zhǎng)，由原來(lái)10min優(yōu)化至3min內(nèi)

（4）完成線上20+個(gè)任務(wù)優(yōu)化及10+個(gè)任務(wù)下線及10+表指標(biāo)下沉，優(yōu)化后節(jié)省任務(wù)耗時(shí)146分鐘，減少CPU損耗800w+，降低內(nèi)存消耗2600w+（相當(dāng)于節(jié)省了8個(gè)200+字段1億數(shù)據(jù)量任務(wù)消耗）

（5）調(diào)度重新分配后2-5點(diǎn)資源使用率由90+%降低至50+%，保障日用資源趨勢(shì)圖無(wú)大突刺波動(dòng)

05小結(jié)

計(jì)算資源治理核心在于降本增效，用有限資源去運(yùn)行更多任務(wù)，通過(guò)一系列治理操作也讓數(shù)倉(cāng)同學(xué)積累技術(shù)經(jīng)驗(yàn)同時(shí)規(guī)范化自身開發(fā)標(biāo)準(zhǔn)，讓治理反推進(jìn)組內(nèi)技術(shù)進(jìn)步。

計(jì)算資源治理是一件長(zhǎng)久之事，并不能因?yàn)橘Y源緊張才去治理，而要將計(jì)算治理常態(tài)化，可通過(guò)周/月資源掃描內(nèi)容及時(shí)推送給每個(gè)同學(xué)，并為之打分，讓每個(gè)任務(wù)都有源可循，有方法可優(yōu)化。

參數(shù)內(nèi)容

參數(shù)并不是設(shè)置越多任務(wù)性能越好，根據(jù)數(shù)據(jù)量、消耗、運(yùn)行時(shí)間進(jìn)行調(diào)整達(dá)到合理效果。

Hive：

（1）set hive.auto.convert.join = true; （是否自動(dòng)轉(zhuǎn)化成Map Join）

（2）set hive.map.aggr=true; （用于控制負(fù)載均衡，頂層的聚合操作放在Map階段執(zhí)行，從而減輕清洗階段數(shù)據(jù)傳輸和Reduce階段的執(zhí)行時(shí)間，提升總體性能，該設(shè)置會(huì)消耗更多的內(nèi)存）

（3）set hive.groupby.skewindata=true; （用于控制負(fù)載均衡，當(dāng)數(shù)據(jù)出現(xiàn)傾斜時(shí)，如果該變量設(shè)置為true，那么Hive會(huì)自動(dòng)進(jìn)行負(fù)載均衡）

（4）set hive.merge.mapfiles=true;  （用于hive引擎合并小文件使用）

（5）set mapreduce.map.memory.mb=4096;      （設(shè)置Map內(nèi)存大小，解決Memory占用過(guò)大/小）

（6）set mapreduce.reduce.memory.mb=4096;（設(shè)置Reduce內(nèi)存大小，解決Memory占用過(guò)大/小）

（7）set hive.exec.dynamic.partition.mode=nonstrict;（動(dòng)態(tài)分區(qū)開啟）

Spark：

（1）set spark.sql.legacy.parquet.datetimeRebaseModeInRead=LEGACY;（用于spark3中字段類型不匹配（例如datetime無(wú)法轉(zhuǎn)換成date），消除sql中時(shí)間歧義，將Spark .sql. LEGACY . timeparserpolicy設(shè)置為L(zhǎng)EGACY來(lái)恢復(fù)Spark 3.0之前的狀態(tài)來(lái)轉(zhuǎn)化）

（2）set spark.sql.adaptive.enabled=true;（是否開啟調(diào)整Partition功能，如果開啟，spark.sql.shuffle.partitions設(shè)置的Partition可能會(huì)被合并到一個(gè)Reducer里運(yùn)行。平臺(tái)默認(rèn)開啟，同時(shí)強(qiáng)烈建議開啟。理由：更好利用單個(gè)Executor的性能，還能緩解小文件問(wèn)題）

（3）set spark.sql.hive.convertInsertingPartitinotallow=false;（解決數(shù)據(jù)無(wú)法同步Impala問(wèn)題，使用Spark3引擎必填）

（4）set spark.sql.finalStage.adaptive.advisoryPartitinotallow=2048M;（Spark小文件合并）?