1. 為什么PB級(jí)Shuffle成為大數(shù)據(jù)處理的性能瓶頸在大規(guī)模數(shù)據(jù)處理場(chǎng)景中Shuffle操作往往成為整個(gè)作業(yè)的性能瓶頸點(diǎn)。以vivo的實(shí)際業(yè)務(wù)為例當(dāng)數(shù)據(jù)規(guī)模達(dá)到PB級(jí)別時(shí)傳統(tǒng)的Shuffle實(shí)現(xiàn)方式會(huì)暴露出三個(gè)致命問題首先是磁盤I/O壓力。在Map階段產(chǎn)生的中間數(shù)據(jù)需要全部落盤Reduce階段再?gòu)拇疟P讀取這個(gè)過程會(huì)產(chǎn)生大量的隨機(jī)讀寫。我們實(shí)測(cè)發(fā)現(xiàn)在100TB級(jí)別的Shuffle場(chǎng)景下磁盤I/O等待時(shí)間占作業(yè)總時(shí)長(zhǎng)的60%以上。其次是網(wǎng)絡(luò)帶寬消耗。所有Mapper節(jié)點(diǎn)需要將數(shù)據(jù)通過網(wǎng)絡(luò)傳輸?shù)絉educer節(jié)點(diǎn)在跨機(jī)房的場(chǎng)景下網(wǎng)絡(luò)延遲和帶寬限制會(huì)導(dǎo)致Shuffle階段耗時(shí)呈指數(shù)級(jí)增長(zhǎng)。一個(gè)典型案例是當(dāng)Shuffle數(shù)據(jù)量從10TB增長(zhǎng)到100TB時(shí)網(wǎng)絡(luò)傳輸時(shí)間從15分鐘激增到3小時(shí)。最后是內(nèi)存管理問題。傳統(tǒng)的Spark Shuffle實(shí)現(xiàn)采用固定大小的內(nèi)存緩沖區(qū)當(dāng)面對(duì)數(shù)據(jù)傾斜嚴(yán)重的場(chǎng)景時(shí)容易導(dǎo)致OOM內(nèi)存溢出。我們?cè)龅揭粋€(gè)極端案例某個(gè)key的數(shù)據(jù)量占整個(gè)Shuffle數(shù)據(jù)量的80%直接導(dǎo)致Executor頻繁崩潰。2. Celeborn架構(gòu)解析vivo的Shuffle優(yōu)化方案Celeborn是vivo自研的分布式Shuffle服務(wù)其核心設(shè)計(jì)理念是將Shuffle數(shù)據(jù)的管理從計(jì)算框架中解耦出來。整個(gè)架構(gòu)包含三個(gè)關(guān)鍵組件2.1 Shuffle Manager服務(wù)層作為中央?yún)f(xié)調(diào)節(jié)點(diǎn)Shuffle Manager負(fù)責(zé)維護(hù)所有Shuffle任務(wù)的元數(shù)據(jù)信息。與原生Spark的實(shí)現(xiàn)不同Celeborn的Manager采用了多級(jí)緩存機(jī)制第一層基于內(nèi)存的元數(shù)據(jù)緩存存儲(chǔ)活躍Shuffle任務(wù)的映射關(guān)系第二層本地SSD持久化存儲(chǔ)保存近24小時(shí)的Shuffle元數(shù)據(jù)第三層分布式KV存儲(chǔ)長(zhǎng)期歸檔歷史任務(wù)信息這種設(shè)計(jì)使得元數(shù)據(jù)查詢延遲從原來的200ms降低到5ms以內(nèi)。2.2 Worker節(jié)點(diǎn)設(shè)計(jì)每個(gè)Worker節(jié)點(diǎn)都部署了三個(gè)核心模塊內(nèi)存池采用動(dòng)態(tài)分配策略根據(jù)當(dāng)前負(fù)載自動(dòng)調(diào)整Map和Reduce任務(wù)的內(nèi)存配額磁盤管理器實(shí)現(xiàn)智能分層存儲(chǔ)熱數(shù)據(jù)存放在NVMe SSD冷數(shù)據(jù)自動(dòng)降級(jí)到HDD網(wǎng)絡(luò)加速模塊內(nèi)置UDP重傳協(xié)議在跨機(jī)房場(chǎng)景下比TCP協(xié)議提升3倍吞吐量我們特別優(yōu)化了小文件合并策略將默認(rèn)的32MB合并閾值調(diào)整為動(dòng)態(tài)算法合并閾值 max(32MB, 總Shuffle數(shù)據(jù)量/1000)這使得在PB級(jí)場(chǎng)景下單個(gè)Shuffle文件大小能自動(dòng)適配到GB級(jí)別。2.3 客戶端SDK為了兼容現(xiàn)有的大數(shù)據(jù)生態(tài)Celeborn提供了多語(yǔ)言客戶端支持Java/Python/Scala原生APISpark/Hadoop/Flink插件Kubernetes Operator實(shí)現(xiàn)自動(dòng)擴(kuò)縮容在vivo內(nèi)部的實(shí)際部署中我們觀察到Spark作業(yè)的Shuffle階段性能提升了4-8倍。特別是在數(shù)據(jù)傾斜嚴(yán)重的廣告推薦場(chǎng)景原先需要6小時(shí)的作業(yè)現(xiàn)在能在45分鐘內(nèi)完成。3. PB級(jí)Shuffle的優(yōu)化實(shí)踐細(xì)節(jié)3.1 動(dòng)態(tài)分區(qū)策略傳統(tǒng)Hash分區(qū)在數(shù)據(jù)傾斜場(chǎng)景下表現(xiàn)糟糕。Celeborn引入了混合分區(qū)算法def selectPartition(key: Any, numPartitions: Int): Int { val hash key.hashCode() if (isHotKey(key)) { // 熱key采用單獨(dú)分區(qū) dedicatedPartition(key) } else { // 普通key使用改良的哈希 (hash Integer.MAX_VALUE) % numPartitions } }配合實(shí)時(shí)監(jiān)控系統(tǒng)當(dāng)檢測(cè)到某個(gè)分區(qū)的數(shù)據(jù)量超過平均值的3倍時(shí)會(huì)自動(dòng)觸發(fā)分裂操作。3.2 零拷貝傳輸優(yōu)化通過Linux內(nèi)核的sendfile系統(tǒng)調(diào)用實(shí)現(xiàn)了磁盤到網(wǎng)卡的直接數(shù)據(jù)傳輸--------------- --------------- | Disk File | | Socket Buffer| -------------- -------------- | | -------- -------- | | -----v--v----- | sendfile() | --------------實(shí)測(cè)顯示這種方案比傳統(tǒng)的用戶態(tài)緩沖拷貝減少30%的CPU使用率。3.3 壓縮算法選型我們對(duì)比了四種壓縮算法在Shuffle場(chǎng)景的表現(xiàn)算法壓縮率CPU消耗適用場(chǎng)景LZ42.1x低網(wǎng)絡(luò)傳輸Zstd3.5x中磁盤存儲(chǔ)Snappy1.8x很低內(nèi)存緩存Gzip4.0x高歸檔存儲(chǔ)最終采用動(dòng)態(tài)策略網(wǎng)絡(luò)傳輸用LZ4磁盤持久化用Zstd level 3內(nèi)存緩存用Snappy。4. 生產(chǎn)環(huán)境部署經(jīng)驗(yàn)4.1 硬件配置建議根據(jù)vivo三個(gè)大數(shù)據(jù)中心的實(shí)測(cè)數(shù)據(jù)我們總結(jié)出以下黃金比例每1TB Shuffle數(shù)據(jù)需要16核CPU64GB內(nèi)存2TB NVMe SSD10Gbps網(wǎng)絡(luò)帶寬4.2 關(guān)鍵參數(shù)調(diào)優(yōu)這些配置在vivo的生產(chǎn)環(huán)境中經(jīng)過驗(yàn)證# Celeborn Worker配置 celeborn.worker.memory.reserve16g # 系統(tǒng)預(yù)留內(nèi)存 celeborn.worker.direct.memory.ratio0.7 # 堆外內(nèi)存占比 celeborn.worker.io.threads32 # IO線程數(shù) # Spark集成配置 spark.shuffle.managerorg.apache.spark.shuffle.celeborn.RssShuffleManager spark.celeborn.master.endpointsceleborn-master:9097 spark.celeborn.push.replicate.enabledtrue # 啟用副本4.3 監(jiān)控指標(biāo)體系建設(shè)我們搭建了完整的監(jiān)控體系重點(diǎn)關(guān)注以下指標(biāo)Shuffle吞吐量GB/s分片均衡度最大/最小分片比磁盤健康度磨損均衡、壞塊率網(wǎng)絡(luò)重傳率1%為健康使用PrometheusGrafana實(shí)現(xiàn)的監(jiān)控看板包含12個(gè)關(guān)鍵圖表能夠?qū)崟r(shí)反映Shuffle服務(wù)狀態(tài)。5. 典型問題排查實(shí)錄5.1 數(shù)據(jù)傾斜場(chǎng)景處理某次推薦算法作業(yè)出現(xiàn)Reduce階段卡頓通過Celeborn的監(jiān)控發(fā)現(xiàn)最大分片數(shù)據(jù)量78GB最小分片數(shù)據(jù)量120MB傾斜比例達(dá)到650:1解決方案分三步實(shí)施在SQL層添加隨機(jī)前綴concat(rand()%100, user_id)調(diào)整Celeborn的動(dòng)態(tài)分區(qū)參數(shù)celeborn.shuffle.detectSkew.enabledtrue celeborn.shuffle.skew.split.threshold5.0在Reduce階段進(jìn)行結(jié)果聚合最終該作業(yè)的執(zhí)行時(shí)間從4.5小時(shí)降至38分鐘。5.2 網(wǎng)絡(luò)閃斷恢復(fù)在跨機(jī)房部署中遇到網(wǎng)絡(luò)抖動(dòng)導(dǎo)致Shuffle失敗我們?cè)鰪?qiáng)了重試機(jī)制指數(shù)退避重試初始間隔100ms最大10s副本自動(dòng)切換當(dāng)主副本失敗時(shí)3秒內(nèi)切換到備用副本黑名單機(jī)制連續(xù)失敗3次的節(jié)點(diǎn)暫停使用5分鐘這些改進(jìn)使得Shuffle服務(wù)的可用性從99.5%提升到99.99%。6. 未來優(yōu)化方向基于當(dāng)前實(shí)踐經(jīng)驗(yàn)我們正在研發(fā)三個(gè)方向的增強(qiáng)功能首先是智能預(yù)取機(jī)制。通過分析作業(yè)DAG圖預(yù)測(cè)下一個(gè)Stage需要的Shuffle數(shù)據(jù)提前加載到內(nèi)存。測(cè)試顯示這可以減少20%的作業(yè)延遲。其次是異構(gòu)硬件支持。正在適配Intel Optane持久內(nèi)存和RDMA網(wǎng)絡(luò)初步測(cè)試顯示在1TB Shuffle場(chǎng)景下延遲降低40%。最后是云原生深度集成?；贙ubernetes的HPAHorizontal Pod Autoscaler實(shí)現(xiàn)自動(dòng)擴(kuò)縮容目標(biāo)是在5分鐘內(nèi)完成從100節(jié)點(diǎn)到1000節(jié)點(diǎn)的擴(kuò)容。