一、現(xiàn)狀和問(wèn)題

1、現(xiàn)狀和問(wèn)題

云音樂(lè)數(shù)倉(cāng)平臺(tái)已經(jīng)上線(xiàn)使用超過(guò)6年時(shí)間，目前累計(jì)用戶(hù)（包括離職人員）超過(guò)700人，每日UV超過(guò)200，涉及數(shù)倉(cāng)開(kāi)發(fā)、數(shù)據(jù)產(chǎn)品、分析師、算法、業(yè)務(wù)開(kāi)發(fā)、QA等幾乎所有角色的開(kāi)發(fā)人員。覆蓋了音樂(lè)所有的業(yè)務(wù)線(xiàn)，一些典型的業(yè)務(wù)類(lèi)型包括索引構(gòu)建、特征開(kāi)發(fā)、內(nèi)容監(jiān)控，以及報(bào)表、線(xiàn)上統(tǒng)計(jì)等。云音樂(lè)業(yè)務(wù)發(fā)展到今天，所有部門(mén)的業(yè)務(wù)都離不開(kāi)大數(shù)據(jù)處理。所有的開(kāi)發(fā)多多少少都會(huì)接觸到大數(shù)據(jù)處理。目前平臺(tái)上實(shí)時(shí)任務(wù)有1600+ ，離線(xiàn)任務(wù)有 7000 到 8000 之間，80% 以上的任務(wù)都是 SQL 任務(wù)。目前整個(gè)云音樂(lè)的集群規(guī)模，純計(jì)算節(jié)點(diǎn)大概有2000+ 臺(tái)機(jī)器，每天原始日志量超過(guò)千億級(jí)別。

2、平臺(tái)思路

平臺(tái)的建設(shè)思路是希望成為連接技術(shù)和業(yè)務(wù)的橋梁，整合技術(shù)和業(yè)務(wù)，通過(guò)平臺(tái)讓數(shù)據(jù)被更高效地用起來(lái)。我們的定位是云音樂(lè)這個(gè)垂直業(yè)務(wù)的數(shù)據(jù)平臺(tái)團(tuán)隊(duì)，需求方更多是音樂(lè)內(nèi)部的需求，并不是通用的集團(tuán)需求，因此與集團(tuán)平臺(tái)或者通用云服務(wù)上數(shù)據(jù)平臺(tái)相比，我們更貼近業(yè)務(wù)，工具也更業(yè)務(wù)化。不同于普適的數(shù)據(jù)開(kāi)發(fā)平臺(tái)更傾向于開(kāi)放通用能力，不會(huì)根據(jù)業(yè)務(wù)的流程規(guī)范做定制，我們需要根據(jù)內(nèi)部的規(guī)范和需求定制化平臺(tái)能力，需要深入到業(yè)務(wù)當(dāng)中去，了解業(yè)務(wù)的需求和開(kāi)發(fā)的痛點(diǎn)，提供整套的解決方案，同時(shí)我們也更關(guān)心業(yè)務(wù)方的成本，希望整體的使用更加經(jīng)濟(jì)，替業(yè)務(wù)省錢(qián)。

3、整體的架構(gòu)

我們整體的能力構(gòu)建于集群服務(wù)之上，集團(tuán)服務(wù)提供了通用的數(shù)據(jù)處理和治理的能能力，比如實(shí)時(shí)任務(wù)開(kāi)發(fā)平臺(tái)sloth，基于Flink，提供了通用的實(shí)時(shí)數(shù)據(jù)處理能力，支持SQL處理實(shí)時(shí)數(shù)據(jù)；離線(xiàn)開(kāi)發(fā)平臺(tái)猛犸提供了通用的離線(xiàn)任務(wù)提交、調(diào)度以及管控的能力，支持MR、SparkSQL、Jar、HiveSQL等多種任務(wù)類(lèi)型；元數(shù)據(jù)中心提供了通用的數(shù)倉(cāng)元數(shù)據(jù)管理能力，血緣追蹤能力；安全中心，基于Ranger提供了通用的權(quán)限管理的基礎(chǔ)能力。在集團(tuán)提供的完善的基礎(chǔ)能力之上，我們根據(jù)云音樂(lè)內(nèi)部的規(guī)范以及需求做了封裝和定制，把業(yè)務(wù)規(guī)范做到平臺(tái)上、整合最佳實(shí)踐，讓用戶(hù)以更低的門(mén)檻和成本，以更高的效率和質(zhì)量在平臺(tái)上完成業(yè)務(wù)需求數(shù)據(jù)處理工作。

目前平臺(tái) 80% 以上的任務(wù)都是平臺(tái)上定制組件來(lái)完成的，對(duì)于平臺(tái)的任務(wù)，我們可以更好地了解任務(wù)的業(yè)務(wù)需求和特點(diǎn)，同時(shí)對(duì)用戶(hù)任務(wù)的管控度也會(huì)比較大，可以更加方便地在用戶(hù)無(wú)感知的情況下批量地進(jìn)行優(yōu)化，實(shí)現(xiàn)開(kāi)發(fā)質(zhì)量的提升，這對(duì)于后面的治理工作是一大助力。當(dāng)然這也是一把雙刃劍，更多的實(shí)現(xiàn)和干預(yù)，也會(huì)帶來(lái)更大的運(yùn)維壓力，對(duì)于團(tuán)隊(duì)人員的開(kāi)發(fā)質(zhì)量和能力也是一大挑戰(zhàn)，需要更加全面地考慮組件的各種應(yīng)用場(chǎng)景。

4、為什么要做這個(gè)治理

我們進(jìn)行治理是基于以下一些原因。

• 第一：去年是降本提效的大年，各大公司都在降本提效，會(huì)有外部的壓力推動(dòng)資源優(yōu)化治理。
• 第二：現(xiàn)狀水位高，超大的業(yè)務(wù)流量導(dǎo)致平臺(tái) Kafka 水位居高不下，長(zhǎng)期處于80%上，很多問(wèn)題比較明顯，突然的一波流量波峰就有可能導(dǎo)致Kafka抖動(dòng)，對(duì)下游的任務(wù)產(chǎn)生影響。
• 第三：音樂(lè)新的平臺(tái)內(nèi)部上線(xiàn)新的埋點(diǎn)體系，新的埋點(diǎn)體系補(bǔ)充了很多業(yè)務(wù)信息，解決了很多統(tǒng)計(jì)問(wèn)題；上報(bào)信息的增多帶來(lái)了三倍的流量增長(zhǎng)，導(dǎo)致Kafka集群以及下游所有Flink任務(wù)的壓力非常大，對(duì)平臺(tái)任務(wù)的穩(wěn)定性會(huì)產(chǎn)生非常大的影響。
• 第四: 云音樂(lè)業(yè)務(wù)發(fā)展到今天，正如前面所說(shuō)，已經(jīng)到了一個(gè)人人用數(shù)據(jù)的狀態(tài)，幾乎所有角色的開(kāi)發(fā)都會(huì)接觸到數(shù)據(jù)開(kāi)發(fā)工作，平臺(tái)大部分用戶(hù)是非專(zhuān)業(yè)數(shù)據(jù)開(kāi)發(fā)，這對(duì)平臺(tái)的穩(wěn)定性、易用性以及運(yùn)維工作都是一個(gè)非常大的挑戰(zhàn)。去年平臺(tái)上60% ~ 70% 的工單問(wèn)題，都是最基礎(chǔ)的性能、概念問(wèn)題，還有使用的配置問(wèn)題，基本上都是通過(guò)文檔，或者簡(jiǎn)單的培訓(xùn)學(xué)習(xí)能夠掌握的問(wèn)題。

二、治理規(guī)劃

治理規(guī)劃主要分為四大塊：

• 第一：摸清現(xiàn)狀

要做哪些事情，現(xiàn)在情況是怎樣的，做到治理有的放矢，快速高效地拿到結(jié)果。

• 第二：運(yùn)動(dòng)式治理

存量歷史任務(wù)多，初期我們需要一些人肉的動(dòng)作來(lái)運(yùn)動(dòng)式地推動(dòng)治理動(dòng)作，快速拿到數(shù)據(jù)結(jié)果，降低整體水位。

• 第三：技術(shù)優(yōu)化

在運(yùn)動(dòng)式治理的過(guò)程中，我們也做了技術(shù)優(yōu)化，來(lái)優(yōu)化任務(wù)的資源使用，提升任務(wù)的穩(wěn)定性，降低整體計(jì)算集群資源的水位、以及Kafka集群的水位。

• 最后：可持續(xù)性

以上三部分事情做完以后，我們還需要考慮治理工作可持續(xù)發(fā)展，不是一錘子買(mǎi)賣(mài)，同時(shí)也不能一直靠堆人力運(yùn)動(dòng)式治理的方法來(lái)解決問(wèn)題，我們希望能夠把運(yùn)動(dòng)式治理的主動(dòng)收益，轉(zhuǎn)化成用戶(hù)主動(dòng)觸發(fā)治理行為的被動(dòng)收益。

1、摸清現(xiàn)狀

為了摸清現(xiàn)狀，我們做了以下一些工作：

和集團(tuán)底層團(tuán)隊(duì)合作，整合集團(tuán)的資源監(jiān)控服務(wù)Smildon ，實(shí)時(shí)獲取集群所有任務(wù)的資源使用情況，統(tǒng)計(jì)所有任務(wù)使用的資源和成本，并將資源和成本直接換算成錢(qián)，通過(guò)前端實(shí)時(shí)的反饋給用戶(hù)。從用戶(hù)角度看，用戶(hù)可以對(duì)任務(wù)的成本有最直接的感知，在使用資源時(shí)會(huì)更加謹(jǐn)慎，同時(shí)在平臺(tái)推進(jìn)用戶(hù)治理時(shí)，用戶(hù)也會(huì)更加配合。從平臺(tái)角度看，可以獲得一個(gè)整體的資源使用大盤(pán)，從資源使用多的任務(wù)開(kāi)始治理，快速收斂資源水位。

同時(shí)我們還收集了任務(wù)并發(fā)和輸入流量的關(guān)系，統(tǒng)計(jì)所有任務(wù)的單位并發(fā)的處理量，然后通過(guò)這個(gè)指標(biāo)快速評(píng)估出平臺(tái)整體的處理能力，通過(guò)這個(gè)值的大小，快速篩選發(fā)現(xiàn)出資源配置可能有問(wèn)題的任務(wù)，高效地進(jìn)行優(yōu)化治理。

為了控制每個(gè)部門(mén)資源的增長(zhǎng)，我們以部門(mén)為單位，整合從Smildon系統(tǒng)收集過(guò)來(lái)的實(shí)時(shí)資源使用數(shù)據(jù)，構(gòu)建邏輯的虛擬隊(duì)列，實(shí)時(shí)地統(tǒng)計(jì)每個(gè)部門(mén)大概用了多少資源，然后劃定初始的限制，如果超過(guò)限制，就要走申請(qǐng)流程來(lái)擴(kuò)容，通過(guò)這種流程上的手段來(lái)控制它的增長(zhǎng)（圖示是一個(gè)虛擬隊(duì)列資源）。

2、高效治理

有了上面說(shuō)的數(shù)據(jù)指標(biāo)就能快速把有問(wèn)題的任務(wù)篩出來(lái)，然后根據(jù)所有任務(wù)的資源的使用量，以及單位并發(fā)的處理能力等指標(biāo)做倒排，快速優(yōu)化治理相關(guān)任務(wù)，收斂資源，拿到數(shù)據(jù)結(jié)果。任務(wù)的治理工作主要分以下幾類(lèi)：

（1）第一：無(wú)用任務(wù)探查下線(xiàn)        

這個(gè)操作的關(guān)鍵在于如何判斷任務(wù)是否還在使用，目前我們的判斷依據(jù)主要有以下幾點(diǎn)：

• 通過(guò)血緣判斷，如果輸出數(shù)據(jù)沒(méi)有任務(wù)消費(fèi)，則大概率任務(wù)是無(wú)用任務(wù)。獲取血緣的手段主要有兩種，對(duì)于SQL任務(wù)和使用我們SDK開(kāi)發(fā)的實(shí)時(shí)任務(wù)，會(huì)通過(guò)靜態(tài)SQL解析獲取任務(wù)的血緣信息，比較準(zhǔn)確。對(duì)于Jar任務(wù)，通過(guò)日志解析抽取關(guān)鍵信息來(lái)獲取血緣，這種方式就不一定能抓取全。所以對(duì)于血緣收集，我們內(nèi)部一直倡導(dǎo)的是約定大于技術(shù)優(yōu)化，推進(jìn)用戶(hù)改造使用SQL或者我們的SDK進(jìn)行開(kāi)發(fā)，來(lái)獲取血緣，而非適配用戶(hù)的開(kāi)發(fā)習(xí)慣，浪費(fèi)人力使用很奇怪的方式去抽取血緣。
• 運(yùn)維積極性，如果任務(wù)長(zhǎng)期無(wú)人運(yùn)維管理，報(bào)警不處理，可以找用戶(hù)確認(rèn)是否還在使用。
• 根據(jù)業(yè)務(wù)周期判斷，如果業(yè)務(wù)已經(jīng)下線(xiàn)，比如日?；顒?dòng)已經(jīng)下線(xiàn)，則可以推進(jìn)相關(guān)任務(wù)下線(xiàn)。

（2）第二：任務(wù)本身的資源不合理

通過(guò)前面提到的任務(wù)的單個(gè)并發(fā)處理數(shù)據(jù)量的指標(biāo)，快速篩選出資源配置不合理可能性比較大的任務(wù)，調(diào)整并發(fā)優(yōu)化資源，由于我們平臺(tái)用戶(hù)大部分都不是數(shù)據(jù)開(kāi)發(fā)的背景，這種Case還是非常多的。

（3）第三：流量萎縮導(dǎo)致任務(wù)資源配置冗余過(guò)剩

還有很多任務(wù)歷史流量很大，但是后來(lái)流量逐步降低，但是任務(wù)資源沒(méi)有相應(yīng)的調(diào)整，也導(dǎo)致了整體資源的配置不合理，這個(gè)后續(xù)可以通過(guò)數(shù)據(jù)來(lái)記錄任務(wù)歷史處理能力，來(lái)判斷整體資源的合理性。

（4）第四：技術(shù)優(yōu)化

為了優(yōu)化整體的資源使用，我們還做了很多技術(shù)優(yōu)化，比如Flink SQL的增強(qiáng)，添加一些額外的能力優(yōu)化整體性能, Kafka寫(xiě)入的優(yōu)化，通過(guò)寫(xiě)入的batch優(yōu)化降低整體Kafka的水位，設(shè)計(jì)開(kāi)發(fā)分區(qū)流表技術(shù)，優(yōu)化流量使用，減少無(wú)用的消息消費(fèi)，降低整體帶寬和整體計(jì)算資源，這些后面會(huì)做詳細(xì)介紹。

3、可持續(xù)性

可持續(xù)發(fā)展指的是讓治理工作變成常態(tài)化，目前這一部分功能還在開(kāi)發(fā)中。我們希望把前文中提到的規(guī)則落到治理平臺(tái)，通過(guò)自動(dòng)化的流程去推進(jìn)用戶(hù)，自動(dòng)掃描出有問(wèn)題的任務(wù)，推進(jìn)通知用戶(hù)，讓用戶(hù)主動(dòng)地做開(kāi)發(fā)治理，讓每個(gè)人都參與到治理工作中來(lái)平臺(tái)方從運(yùn)動(dòng)式治理的主動(dòng)收益，變成自動(dòng)化的被動(dòng)收益。

三、技術(shù)優(yōu)化

接下來(lái)將分三部分來(lái)介紹技術(shù)上的優(yōu)化：Flink SQL優(yōu)化、Kafka 的batch優(yōu)化，以及我們?cè)O(shè)計(jì)開(kāi)發(fā)的“分區(qū)流表”的優(yōu)化。

1、Flink SQL 優(yōu)化

Flink SQL的發(fā)布，大大降低了實(shí)時(shí)計(jì)算的開(kāi)發(fā)門(mén)檻，提升了開(kāi)發(fā)效率，但是它也帶來(lái)了一些問(wèn)題，SQL背后邏輯的不透明，讓用戶(hù)能夠控制的東西也變少了，這導(dǎo)致了一些不必要的計(jì)算邏輯，同時(shí)用戶(hù)能做的優(yōu)化手段也變少了，這導(dǎo)致了中間有很多的資源浪費(fèi)。下面我們通過(guò)一些Case來(lái)說(shuō)明。

（1）Case 1 :  消息反序列化前置優(yōu)化

背景：日志消息格式為userid\001os\001action\001logtime\001props。Props 是 JSON格式，所以在讀取流表的過(guò)程中大部分性能損耗都在JSON 的解析上。離線(xiàn)的場(chǎng)景下我們可以做列裁剪，只讀需要的數(shù)據(jù)，但是在實(shí)時(shí)情況下，還沒(méi)有那么成熟，不管我們需要不需要props這個(gè)字段，F(xiàn)linkSQL本身都會(huì)對(duì)整條消息做解析，這導(dǎo)致了很多的資源浪費(fèi)。

為了解決這個(gè)問(wèn)題，我在反序列化上做了一些優(yōu)化，用戶(hù)可以通過(guò)一些配置在解析完整日志前做一些過(guò)濾，比如上圖中的這兩條SQL的對(duì)比，在解析整條消息之前，通過(guò)keyword配置做關(guān)鍵字過(guò)濾，將不包含‘user-fm’關(guān)鍵字的消息全部過(guò)濾掉，在解析props之前，通過(guò)os.list和action.list過(guò)濾掉多余的消息，通過(guò)這些配置可以減少大量無(wú)用消息的解析，大大提升整體任務(wù)的性能，降低CPU的消耗。這些優(yōu)化在很多情況下效果非常明顯，極端情況下能減少50% 以上的性能損耗。

與離線(xiàn)場(chǎng)景下的列裁剪類(lèi)似，按需解析，按需反序列化，這個(gè)優(yōu)化還可以持續(xù)優(yōu)化，現(xiàn)在還需要用戶(hù)手動(dòng)去配置，我們最終的目標(biāo)是根據(jù)用戶(hù)的 Select 字段結(jié)合format的實(shí)現(xiàn)做自動(dòng)的列裁剪優(yōu)化。

（2）Case 2 :索引構(gòu)建場(chǎng)景

第二個(gè)case是索引構(gòu)建場(chǎng)景。很多索引是通過(guò)關(guān)聯(lián)多張數(shù)據(jù)庫(kù)表，生成大寬表，寫(xiě)入到索引引擎里面的，然后提供給前端用戶(hù)去查詢(xún)。大致流程為，用戶(hù)通過(guò)Flink訂閱數(shù)據(jù)庫(kù)的binlog日志來(lái)監(jiān)聽(tīng)業(yè)務(wù)庫(kù)的數(shù)據(jù)變化，然后把binlog里面關(guān)鍵數(shù)據(jù)和很多業(yè)務(wù)DB表進(jìn)行關(guān)聯(lián)，生成一個(gè)大寬表，最后再通過(guò)Flink寫(xiě)入到索引構(gòu)建引擎里面，供用戶(hù)查詢(xún)。這里存在幾個(gè)問(wèn)題：

• 第一：Flink SQL讀 Kafka受到Kafka 分區(qū)的限制，比如10個(gè)分區(qū)只能通過(guò)10個(gè)并發(fā)去讀取和消費(fèi)；
• 第二：當(dāng)維表關(guān)聯(lián)特別多的時(shí)候，因?yàn)樯嫌蔚姆謪^(qū)有限，下游維表關(guān)聯(lián)受到維表查詢(xún)性能的限制，表越多單條消息的處理性能越差。

兩者結(jié)合就導(dǎo)致整體的處理性能無(wú)法做到水平擴(kuò)展，無(wú)論怎樣擴(kuò)大Flink任務(wù)并發(fā)，始終也都只有10個(gè)并發(fā)在處理消息，導(dǎo)致任務(wù)的延遲非常嚴(yán)重。

我們的優(yōu)化方案是：

• 第一：完善 Metrics 監(jiān)控，把所有的維表關(guān)聯(lián)的 Metrics，比如每張維表查詢(xún)的RT，消息反序列化的性能、以及寫(xiě)入三方存儲(chǔ)的RT 相關(guān)監(jiān)控指標(biāo)全部收集上來(lái)，寫(xiě)入到任務(wù)的監(jiān)控里面去，在Granafa上展示出來(lái)，這樣如果哪張維表因?yàn)樗饕O(shè)計(jì)不當(dāng)，導(dǎo)致維表關(guān)聯(lián)的性能特別差，就可以通過(guò) Granafa的監(jiān)控頁(yè)面快速發(fā)現(xiàn)，并進(jìn)行優(yōu)化。
• 第二：針對(duì)維表越多性能越差的問(wèn)題，添加異步關(guān)聯(lián)配置，開(kāi)啟 Flink  AyncIO的特性，通過(guò)異步關(guān)聯(lián)的方式提升任務(wù)整體的處理能力。
• 第三：關(guān)于處理能力受Kafka分區(qū)限制的問(wèn)題：Flink在讀取Kafka消息時(shí)會(huì)自動(dòng)做OperationChain的優(yōu)化，會(huì)把讀取動(dòng)作、解析動(dòng)作、維外關(guān)聯(lián)寫(xiě)入動(dòng)作全部綁定在一起，導(dǎo)致這一系列動(dòng)作都會(huì)受到Kafka 的并發(fā)限制，整個(gè)處理能力非常糟糕。特別是在維表關(guān)聯(lián)特別多的時(shí)候，即使開(kāi)啟異步優(yōu)化，整體性能的提升也不是特別明顯，這個(gè)時(shí)候需要一個(gè)能力把行為拆開(kāi)，分別設(shè)置并發(fā)。所以在Flink SQL 里面加了一個(gè)配置，在讀取表消息的過(guò)程中添加一個(gè)修改并發(fā)的操作，把讀取消息行為和后續(xù)的解析處理消息的行為拆開(kāi)。通過(guò)在中間加一個(gè)rescale或者rebalance的操作，分別設(shè)置讀取和后續(xù)解析處理的并發(fā)。在沒(méi)有按照消息內(nèi)容shard的需求時(shí)，我們推薦 rescale，因?yàn)?rescale 的性能損耗比較小。這樣后續(xù)維表關(guān)聯(lián)的功能就可以不受Kafka分區(qū)數(shù)量的限制，可以通過(guò)調(diào)整后續(xù)處理的并發(fā)，做到處理能力水平擴(kuò)展，當(dāng)然中間添加rescale或者rebalance操作會(huì)導(dǎo)致消息亂序，在對(duì)消息順序有要求的場(chǎng)景上時(shí)，這個(gè)優(yōu)化不能使用。
• 最后，這種類(lèi)型的任務(wù)都是IO密集型任務(wù)，輸入流量往往都不是很大，加并發(fā)只是為了增加DB維表查詢(xún)的并發(fā)，提升任務(wù)整體的吞吐。所以在優(yōu)化這種類(lèi)型時(shí)，我們還會(huì)優(yōu)化CPU的配置，通過(guò)yarn.containers.vcors的配置，做細(xì)粒度CPU資源分配。默認(rèn)情況下一個(gè)slot分配一個(gè)CPU，通過(guò)這個(gè)配置可以控制比例，比如4個(gè)Slot， yarn.containers.vcors 配置為2的話(huà)，一個(gè)Slot 就只分配到0.5個(gè)CPU,同時(shí)也能帶來(lái)資源的節(jié)省。

2、Kafka 的Batch 優(yōu)化

前面已經(jīng)提到，我們的Kafka集群一直處于水位比較高的狀態(tài)，高峰期水位達(dá)到80%，加上業(yè)務(wù)即將上線(xiàn)新的埋點(diǎn)體系，會(huì)帶來(lái)三倍的流量增長(zhǎng)。為了降低整體Kafka集群水位，我們做了以下工作：

• 第一 完善Kafka監(jiān)控

早期我們這邊Kafka的運(yùn)維體系比較簡(jiǎn)陋，監(jiān)控指標(biāo)不夠完善，導(dǎo)致我們的優(yōu)化工作難以下手，為了更好地了解Kafka水位高的原因，我們參考了 Kafka 社區(qū)的方案搭建了非常完善的監(jiān)控，獲得了比較完善的數(shù)據(jù)監(jiān)控，給我們的優(yōu)化提供了方向，我們通過(guò)監(jiān)控的數(shù)據(jù)發(fā)現(xiàn)了下面提到的問(wèn)題。

• 第二 流量均衡的問(wèn)題

一個(gè)Kakfa集群服務(wù)很多業(yè)務(wù)，每個(gè)業(yè)務(wù)有很多消息隊(duì)列topic，每個(gè)topic又有很多partition分區(qū)，這些分區(qū)分布在集群的機(jī)器上，分區(qū)和機(jī)器的關(guān)系都是手動(dòng)維護(hù)的，消息分區(qū)的分布不均，每個(gè)分區(qū)消息的流量大小也不相同，這直接導(dǎo)致了有些機(jī)器的負(fù)載比較高，有些機(jī)器的負(fù)載低。這塊目前的優(yōu)化方案比較簡(jiǎn)單粗暴，通過(guò)監(jiān)控直觀地看到每臺(tái)機(jī)器的流量情況，然后PE通過(guò)工具手動(dòng)調(diào)整topic的partition分布情況，保障每臺(tái)機(jī)器流量均衡穩(wěn)定。這個(gè)問(wèn)題也是Kafka開(kāi)源版本普遍存在的問(wèn)題。未來(lái)會(huì)考慮把Kafka 替換成 Pulsar，通過(guò)存算分離的架構(gòu)優(yōu)勢(shì)來(lái)解決這個(gè)問(wèn)題。

• 第三 消息發(fā)送的優(yōu)化

通過(guò)監(jiān)控發(fā)現(xiàn)，以前的Kafka的水位高大部分是因?yàn)樘幚砭€(xiàn)程池不夠以及磁盤(pán)的 IO比較大，但是整體消息量還好。深挖后發(fā)現(xiàn)，消息發(fā)送的batch size配置沒(méi)有生效，很多時(shí)候一次發(fā)送只有一到兩條消息，這樣100條消息就要發(fā)送100次請(qǐng)求，這會(huì)導(dǎo)致Kafka消息處理的線(xiàn)程水位非常高，磁盤(pán)IO的操作頻次也是一樣。但是為什么batch size的配置沒(méi)有生效？調(diào)研發(fā)現(xiàn)Kafka batch 與batch size 大小的配置、 以及l(fā)iner.ms最大容忍延遲時(shí)間這兩個(gè)配置有關(guān)，liner.ms的默認(rèn)配置為0，但是當(dāng)我們優(yōu)化了這個(gè)配置以后，batch效果還是不明顯，最后我們發(fā)現(xiàn)還與producer partitioner的策略有關(guān)系。

Partitioner優(yōu)化策略考慮以下幾點(diǎn)：

• 分區(qū)均衡，要保障消息均勻地發(fā)送到所有partition當(dāng)中，不然會(huì)導(dǎo)致數(shù)據(jù)傾斜問(wèn)題，對(duì)機(jī)器以及下游消費(fèi)者都會(huì)帶來(lái)壓力。
• 單個(gè)消息體不能太大，不然消息的延遲會(huì)增加，下游處理單條消息時(shí)壓力也會(huì)變大。
• 最大容忍時(shí)間：當(dāng)消息體在最大容忍時(shí)間還沒(méi)有積累到配置的最大size時(shí)，也會(huì)觸發(fā)發(fā)送請(qǐng)求。

在三者之間要做 trade Off，太大的話(huà)會(huì)影響延遲，太小的話(huà)整個(gè) IO 也會(huì)不行。Kafka在2.4版本里引入了新的 Partition策略：Sticky Partitioner ，在公共的Partitioner接口中添加了一個(gè)新的 onNewBatch 方法，每次創(chuàng)建新的 Batch的時(shí)候會(huì)調(diào)這個(gè)方法，在這個(gè)方法調(diào)用時(shí)， Sticky Partitioner 會(huì)隨機(jī)選擇一個(gè)分區(qū)，然后將所有的消息都會(huì)放到一個(gè)Cache 當(dāng)中，在下一次OnNewBatch 的時(shí)候會(huì)把Cache中所有消息打包成一個(gè)batch發(fā)送到隨機(jī)選擇的那個(gè)分區(qū)當(dāng)中，下一次再隨機(jī)選一個(gè)分區(qū)，繼續(xù)積累消息到Cache中然后打包發(fā)送，這種策略既保證了分區(qū)的均衡性又保證了Batch效果的最大化。經(jīng)過(guò)實(shí)踐測(cè)試，通過(guò)Sticky Partitioner的Batch策略帶來(lái)的性能優(yōu)化非常明顯，整個(gè)Kafka集群的水位從80%降低到了30%。

3、分區(qū)流表優(yōu)化

（1）數(shù)倉(cāng)的處理流程

在離線(xiàn)場(chǎng)景下，我們可以通過(guò)列存儲(chǔ)、分桶、分區(qū)、索引等諸多手段來(lái)減少不必要的數(shù)據(jù)讀取，從而提升程序的性能。為了降低整體實(shí)時(shí)集群的使用成本，扛住新埋點(diǎn)三倍的流量沖擊，我們參考了 Hive 的分區(qū)設(shè)計(jì)，實(shí)現(xiàn)了一套實(shí)時(shí)分區(qū)流表的設(shè)計(jì)。

上圖是一個(gè)比較常規(guī)的實(shí)時(shí)數(shù)倉(cāng)的處理流程圖，正常的日志處理流程包括 DS 歸檔（網(wǎng)易內(nèi)部服務(wù)，收集日志到Kafka和HDFS），然后通過(guò)Flink清洗格式化到ODS層，再到DWD層，業(yè)務(wù)應(yīng)用程序消費(fèi)DWD，生產(chǎn)出ADS層給上層應(yīng)用提供服務(wù)。整個(gè)流程中，ODS 的日志量非常大，在開(kāi)發(fā)DWD的時(shí)候，需要消費(fèi)全量的ODS層的日志，如果ODS層日志流量大小是700M/S，那么下游所有的DWD的任務(wù)都要消費(fèi)這700M/S的流量，要處理這么大的流量，大概要900Core資源，相當(dāng)于9臺(tái)最新配置的物理服務(wù)器，每多一個(gè)DWD表任務(wù)，就需要增加9臺(tái)物理服務(wù)器。另外在這么大流量的情況下，任務(wù)的穩(wěn)定性無(wú)法保障，任何一波日志的波動(dòng)，都會(huì)對(duì)下游任務(wù)產(chǎn)生比較大的影響，Kafka的壓力也會(huì)非常大，成本也無(wú)法接受。

（2）歷史方案

我們以前的方案是在源頭上對(duì)原始日志做拆分，通過(guò)單獨(dú)分發(fā)程序，按業(yè)務(wù)需求將原始日志拆分成不同的Topic，業(yè)內(nèi)一些公司也是也是這么做的，但是這種做法有如下問(wèn)題：

① 運(yùn)維成本高，拆分粒度相對(duì)較粗，下游還是有一定程度的流量的無(wú)用消費(fèi)，后期拆分也比較困難。

② 用戶(hù)在使用實(shí)時(shí)流表的時(shí)候需要很多的先驗(yàn)知識(shí)，需要了解消息的分發(fā)規(guī)則讀取正確的流表，使用成本高。

③ 實(shí)時(shí)數(shù)倉(cāng)的建模和離線(xiàn)不能統(tǒng)一，后期如果想做批流一體，實(shí)時(shí)數(shù)倉(cāng)和離線(xiàn)數(shù)倉(cāng)沒(méi)法做到Schema的一致，繼而也沒(méi)法做到一套代碼同時(shí)支持實(shí)時(shí)和離線(xiàn)。

④ 無(wú)法遷移復(fù)用，在源頭單獨(dú)做一個(gè)定制的分發(fā)程序，下游消費(fèi)的分發(fā)后數(shù)據(jù)有可能存在流量大的問(wèn)題，下游用戶(hù)沒(méi)法輕松的復(fù)用這套方案，不能夠持續(xù)產(chǎn)生價(jià)值。

（3）分區(qū)流表優(yōu)化

我們參考Hive表的分區(qū)設(shè)計(jì)，重新設(shè)計(jì)了實(shí)時(shí)流表的元數(shù)據(jù)結(jié)果，讓實(shí)時(shí)流表也有分區(qū)的概念，分區(qū)元信息中包含分區(qū)和Kafka topic的映射關(guān)系，然后我們定制修改了Kafka Connector，根據(jù)流表的分區(qū)元信息，寫(xiě)入流表時(shí)根據(jù)消息中分區(qū)字段的內(nèi)容和分區(qū)的元信息將消息寫(xiě)入到不同的topic當(dāng)中。讀取流表時(shí)，我們?cè)贙afka Connector的基礎(chǔ)上實(shí)現(xiàn)了分區(qū)下推，會(huì)根據(jù)用戶(hù)查詢(xún)SQL中分區(qū)條件，自動(dòng)推斷出需要的分區(qū)topic，裁剪掉不需要的分區(qū)topic，減少不必要的消息讀取，減少資源浪費(fèi)。

有了分區(qū)流表以后，我們就可以把DS歸檔過(guò)來(lái)的日志，下游所有的DWD任務(wù)都能夠在無(wú)感知的情況下享受到分區(qū)流表帶來(lái)的流量?jī)?yōu)化的紅利，用戶(hù)不需要關(guān)注太多，定好分區(qū)規(guī)則，使用SQL 讀寫(xiě)，不需要構(gòu)建單獨(dú)的程序做分發(fā)，復(fù)用的成本非常低，下游的大流量的 DWD 層表的建設(shè)也可以以很低的成本復(fù)用分區(qū)流表技術(shù)，做到整體流量的減少。

示例中，只需要定義好寫(xiě)分區(qū)字段，會(huì)根據(jù)分區(qū)字段自動(dòng)寫(xiě)到相應(yīng)的 topic 里，讀取這個(gè)分區(qū)字段的查詢(xún)條件，會(huì)自動(dòng)推斷出 topic 來(lái)源。

四、未來(lái)規(guī)劃

未來(lái)規(guī)劃，主要是兩塊：大數(shù)據(jù)容器化改造和自動(dòng)化治理平臺(tái)。

1、容器化改造

我們希望通過(guò)容器化改造獲得以下幾方面的能力：

• 第一：優(yōu)秀的資源隔離能力，通過(guò)K8S的容器化CGroup比較好的資源隔離能力，解決Yarn環(huán)境下任務(wù)之間相互影響的問(wèn)題，雖然Yarn上也可以開(kāi)啟CGroup，但是配置起來(lái)非常不靈活，運(yùn)維起來(lái)比較困難。   
• 第二：精細(xì)化資源配置

在Yarn環(huán)境下，我們只能通過(guò)yarn.containers.vcores做一些資源精細(xì)化的資源調(diào)整，但是整體粒度較粗，在K8S上可以做到1/1000Core粒度，資源配置更加精細(xì)化，更加靈活。

• 第三：宏觀監(jiān)控體系

在Yarn環(huán)境下，因?yàn)闆](méi)有很好的資源隔離性，本身也缺乏container級(jí)別的資源利用率的詳細(xì)指標(biāo)，導(dǎo)致我們很難從機(jī)器負(fù)載、CPU利用率、內(nèi)存利用，帶寬使用等宏觀指標(biāo)來(lái)評(píng)估任務(wù)資源使用的合理性；在K8S環(huán)境下，擁有比較好的資源隔離性，以及完善資源指標(biāo)，我們搭建任務(wù)級(jí)別的宏觀監(jiān)控體系，參考一般web應(yīng)用程序，通過(guò)CPU 利用率、 IO 利用率、內(nèi)存利用率等宏觀監(jiān)控快速評(píng)估出Flink應(yīng)用的資源申請(qǐng)的合理性，快速治理優(yōu)化。

• 第四：靈活的資源調(diào)度能力

 K8S本身可以定制非常靈活的調(diào)度策略，方便我們根據(jù)任務(wù)特點(diǎn)選擇不同類(lèi)型的機(jī)器；還可以和其它類(lèi)型業(yè)務(wù)（如機(jī)器學(xué)習(xí)、在線(xiàn)業(yè)務(wù)、離線(xiàn)計(jì)算等）的資源混合部署，達(dá)到整體資源利用率的最大化。

2、自動(dòng)化治理平臺(tái)

正如前面提到的，我們希望通過(guò)收集數(shù)倉(cāng)、任務(wù)、用戶(hù)平臺(tái)要素的元數(shù)據(jù)，構(gòu)建元數(shù)據(jù)倉(cāng)庫(kù)，在元數(shù)據(jù)倉(cāng)庫(kù)的基礎(chǔ)之上，使用這些元數(shù)據(jù)進(jìn)行規(guī)則配置。在開(kāi)發(fā)上線(xiàn)前，通過(guò)規(guī)則進(jìn)行一些合法性的校驗(yàn)，如SQL是否規(guī)范、報(bào)警是否完備等前置檢查；服務(wù)中通過(guò)規(guī)則定期掃描，自動(dòng)發(fā)現(xiàn)問(wèn)題，如資源是否合理、是否可以下線(xiàn)等，自動(dòng)化地推進(jìn)用戶(hù)進(jìn)行治理；任務(wù)治理以后回收治理效果，通過(guò)曬治理結(jié)果形成紅黑榜，形成一個(gè)良性的閉環(huán)。

五、Q&A

Q1：有基于分區(qū)流表做一些流批一體的工作嗎？

A：我們現(xiàn)在的方案都是通過(guò)構(gòu)建一層數(shù)據(jù)模型層，數(shù)據(jù)模型層會(huì)關(guān)聯(lián)離線(xiàn)數(shù)倉(cāng)表和實(shí)時(shí)數(shù)倉(cāng)表，在離線(xiàn)場(chǎng)景讀取離線(xiàn)表，實(shí)時(shí)場(chǎng)景讀取實(shí)時(shí)流表。前面提到的分區(qū)流表的技術(shù)解決了實(shí)時(shí)數(shù)倉(cāng)表單表流量過(guò)大的問(wèn)題，做到了離線(xiàn)數(shù)倉(cāng)和實(shí)時(shí)數(shù)倉(cāng)建模的統(tǒng)一，所以在數(shù)據(jù)模型這一層就很容易做到了統(tǒng)一的映射。

文中提到了一個(gè) FastX 的開(kāi)發(fā)工具，是一個(gè)基于數(shù)據(jù)模型的低代碼的開(kāi)發(fā)工作，可以在FastX管理數(shù)據(jù)模型，然后在數(shù)據(jù)模型的基礎(chǔ)上通過(guò)低代碼的方式做配置化的開(kāi)發(fā)。通過(guò)低代碼的方式生成一套統(tǒng)一的計(jì)算邏輯，實(shí)現(xiàn)一套邏輯配置，在流批兩套環(huán)境下運(yùn)行。

Q2：怎么屏蔽SQL的不同

A：FastX會(huì)通過(guò)低代碼配置化的方式生產(chǎn)一套統(tǒng)一的DSL，在實(shí)時(shí)場(chǎng)景下會(huì)選擇實(shí)時(shí)數(shù)倉(cāng)的數(shù)據(jù)源，然后將DSL生成實(shí)時(shí)的FlinkSQL，離線(xiàn)情況下選擇離線(xiàn)數(shù)據(jù)源生成Spark SQL進(jìn)行執(zhí)行。上層交互是有限的，整體的算子也是可控的，所以我們逐步覆蓋業(yè)務(wù)場(chǎng)景，實(shí)現(xiàn)一套邏輯，同時(shí)跑在離線(xiàn)和實(shí)時(shí)環(huán)境上。目前FastX定位是一個(gè)基于業(yè)務(wù)場(chǎng)景的開(kāi)發(fā)平臺(tái)，做全場(chǎng)景的肯定很難，我們希望根據(jù)二八原則，能夠覆蓋80%的業(yè)務(wù)場(chǎng)景。

Q3：實(shí)時(shí)數(shù)倉(cāng)治理跟離線(xiàn)數(shù)倉(cāng)治理在方法論上有哪些異同點(diǎn)？

A：方法論上感覺(jué)是差不多的，但是實(shí)時(shí)數(shù)倉(cāng)與離線(xiàn)數(shù)倉(cāng)相比，發(fā)展時(shí)間還比較短。離線(xiàn)數(shù)倉(cāng)治理上有很多數(shù)據(jù)指標(biāo)來(lái)評(píng)價(jià)數(shù)倉(cāng)的質(zhì)量，比如穿透率、復(fù)用率、閑置濾等來(lái)評(píng)價(jià)數(shù)據(jù)資產(chǎn)的好壞，會(huì)朝著這些量化的目標(biāo)來(lái)做數(shù)據(jù)倉(cāng)庫(kù)結(jié)構(gòu)上的優(yōu)化。在實(shí)時(shí)場(chǎng)景下，數(shù)倉(cāng)的結(jié)構(gòu)還比較簡(jiǎn)單，層次比較少，一般不會(huì)在實(shí)時(shí)場(chǎng)景下建 DWS 層，頂多是 ODS 層到DWD 層，然后再到業(yè)務(wù)層，構(gòu)建 DWS 層成本也非常高，存儲(chǔ)也很難選。本身實(shí)時(shí)場(chǎng)景的Flink任務(wù)對(duì)資源以及穩(wěn)定性、延遲等都比較敏感，必要的時(shí)候，數(shù)倉(cāng)建設(shè)的規(guī)范需要給資源性能讓步，所以在實(shí)時(shí)數(shù)倉(cāng)的治理上我們更關(guān)注穩(wěn)定性、資源的治理。