網(wǎng)易云音樂實時數(shù)倉治理優(yōu)化實踐

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

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

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

　　2、平臺思路

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

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

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

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

　　4、為什么要做這個治理

　　我們進行治理是基于以下一些原因。

　　第一：去年是降本提效的大年，各大公司都在降本提效，會有外部的壓力推動資源優(yōu)化治理。

　　第二：現(xiàn)狀水位高，超大的業(yè)務(wù)流量導致平臺 Kafka 水位居高不下，長期處于80%上，很多問題比較明顯，突然的一波流量波峰就有可能導致Kafka抖動，對下游的任務(wù)產(chǎn)生影響。

　　第三：音樂新的平臺內(nèi)部上線新的埋點體系，新的埋點體系補充了很多業(yè)務(wù)信息，解決了很多統(tǒng)計問題;上報信息的增多帶來了三倍的流量增長，導致Kafka集群以及下游所有Flink任務(wù)的壓力非常大，對平臺任務(wù)的穩(wěn)定性會產(chǎn)生非常大的影響。

　　第四: 云音樂業(yè)務(wù)發(fā)展到今天，正如前面所說，已經(jīng)到了一個人人用數(shù)據(jù)的狀態(tài)，幾乎所有角色的開發(fā)都會接觸到數(shù)據(jù)開發(fā)工作，平臺大部分用戶是非專業(yè)數(shù)據(jù)開發(fā)，這對平臺的穩(wěn)定性、易用性以及運維工作都是一個非常大的挑戰(zhàn)。去年平臺上60% ~ 70% 的工單問題，都是最基礎(chǔ)的性能、概念問題，還有使用的配置問題，基本上都是通過文檔，或者簡單的培訓學習能夠掌握的問題。

　　二、治理規(guī)劃

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

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

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

　　第二：運動式治理

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

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

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

　　最后：可持續(xù)性

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

　　1、摸清現(xiàn)狀

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

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

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

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

　　2、高效治理

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

　　(1)第一：無用任務(wù)探查下線

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

　　通過血緣判斷，如果輸出數(shù)據(jù)沒有任務(wù)消費，則大概率任務(wù)是無用任務(wù)。獲取血緣的手段主要有兩種，對于SQL任務(wù)和使用我們SDK開發(fā)的實時任務(wù)，會通過靜態(tài)SQL解析獲取任務(wù)的血緣信息，比較準確。對于Jar任務(wù)，通過日志解析抽取關(guān)鍵信息來獲取血緣，這種方式就不一定能抓取全。所以對于血緣收集，我們內(nèi)部一直倡導的是約定大于技術(shù)優(yōu)化，推進用戶改造使用SQL或者我們的SDK進行開發(fā)，來獲取血緣，而非適配用戶的開發(fā)習慣，浪費人力使用很奇怪的方式去抽取血緣。

　　運維積極性，如果任務(wù)長期無人運維管理，報警不處理，可以找用戶確認是否還在使用。

　　根據(jù)業(yè)務(wù)周期判斷，如果業(yè)務(wù)已經(jīng)下線，比如日常活動已經(jīng)下線，則可以推進相關(guān)任務(wù)下線。

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

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

　　(3)第三：流量萎縮導致任務(wù)資源配置冗余過剩

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

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

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

　　3、可持續(xù)性

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

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

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

　　1、Flink SQL 優(yōu)化

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

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

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

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

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

　　(2)Case 2 :索引構(gòu)建場景

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

　　第一：Flink SQL讀 Kafka受到Kafka 分區(qū)的限制，比如10個分區(qū)只能通過10個并發(fā)去讀取和消費;

　　第二：當維表關(guān)聯(lián)特別多的時候，因為上游的分區(qū)有限，下游維表關(guān)聯(lián)受到維表查詢性能的限制，表越多單條消息的處理性能越差。

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

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

　　第一：完善 Metrics 監(jiān)控，把所有的維表關(guān)聯(lián)的 Metrics，比如每張維表查詢的RT，消息反序列化的性能、以及寫入三方存儲的RT 相關(guān)監(jiān)控指標全部收集上來，寫入到任務(wù)的監(jiān)控里面去，在Granafa上展示出來，這樣如果哪張維表因為索引設(shè)計不當，導致維表關(guān)聯(lián)的性能特別差，就可以通過 Granafa的監(jiān)控頁面快速發(fā)現(xiàn)，并進行優(yōu)化。

　　第二：針對維表越多性能越差的問題，添加異步關(guān)聯(lián)配置，開啟 Flink AyncIO的特性，通過異步關(guān)聯(lián)的方式提升任務(wù)整體的處理能力。

　　第三：關(guān)于處理能力受Kafka分區(qū)限制的問題：Flink在讀取Kafka消息時會自動做OperationChain的優(yōu)化，會把讀取動作、解析動作、維外關(guān)聯(lián)寫入動作全部綁定在一起，導致這一系列動作都會受到Kafka 的并發(fā)限制，整個處理能力非常糟糕。特別是在維表關(guān)聯(lián)特別多的時候，即使開啟異步優(yōu)化，整體性能的提升也不是特別明顯，這個時候需要一個能力把行為拆開，分別設(shè)置并發(fā)。所以在Flink SQL 里面加了一個配置，在讀取表消息的過程中添加一個修改并發(fā)的操作，把讀取消息行為和后續(xù)的解析處理消息的行為拆開。通過在中間加一個rescale或者rebalance的操作，分別設(shè)置讀取和后續(xù)解析處理的并發(fā)。在沒有按照消息內(nèi)容shard的需求時，我們推薦 rescale，因為 rescale 的性能損耗比較小。這樣后續(xù)維表關(guān)聯(lián)的功能就可以不受Kafka分區(qū)數(shù)量的限制，可以通過調(diào)整后續(xù)處理的并發(fā)，做到處理能力水平擴展，當然中間添加rescale或者rebalance操作會導致消息亂序，在對消息順序有要求的場景上時，這個優(yōu)化不能使用。

　　最后，這種類型的任務(wù)都是IO密集型任務(wù)，輸入流量往往都不是很大，加并發(fā)只是為了增加DB維表查詢的并發(fā)，提升任務(wù)整體的吞吐。所以在優(yōu)化這種類型時，我們還會優(yōu)化CPU的配置，通過yarn.containers.vcors的配置，做細粒度CPU資源分配。默認情況下一個slot分配一個CPU，通過這個配置可以控制比例，比如4個Slot， yarn.containers.vcors 配置為2的話，一個Slot 就只分配到0.5個CPU,同時也能帶來資源的節(jié)省。

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

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

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

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

　　第二 流量均衡的問題

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

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

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

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

　　分區(qū)均衡，要保障消息均勻地發(fā)送到所有partition當中，不然會導致數(shù)據(jù)傾斜問題，對機器以及下游消費者都會帶來壓力。

　　單個消息體不能太大，不然消息的延遲會增加，下游處理單條消息時壓力也會變大。

　　最大容忍時間：當消息體在最大容忍時間還沒有積累到配置的最大size時，也會觸發(fā)發(fā)送請求。

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

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

　　(1)數(shù)倉的處理流程

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

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

　　(2)歷史方案

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

　�、� 運維成本高，拆分粒度相對較粗，下游還是有一定程度的流量的無用消費，后期拆分也比較困難。

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

　�、� 實時數(shù)倉的建模和離線不能統(tǒng)一，后期如果想做批流一體，實時數(shù)倉和離線數(shù)倉沒法做到Schema的一致，繼而也沒法做到一套代碼同時支持實時和離線。

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

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

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

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

　　示例中，只需要定義好寫分區(qū)字段，會根據(jù)分區(qū)字段自動寫到相應(yīng)的 topic 里，讀取這個分區(qū)字段的查詢條件，會自動推斷出 topic 來源。

　　四、未來規(guī)劃

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

　　1、容器化改造

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

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

　　第二：精細化資源配置

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

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

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

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

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

　　2、自動化治理平臺

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