芒果 TV 基于 Flink 的實時數(shù)倉建設(shè)實踐與演進

01

芒果 TV 實時數(shù)倉建設(shè)歷程

芒果 TV 實時數(shù)倉的建設(shè)共分為三個階段，14-19 年為第一階段，技術(shù)選型采用 Storm/Flink Java+Spark SQL。20-22 年上半年為第二階段，技術(shù)選型采用 Flink SQL+Spark SQL 。22 年下半年-至今為第三階段，技術(shù)選型采用 Flink SQL+StarRocks。每一次升級都是在原有基礎(chǔ)上進行迭代，以求更全面的功能，更快的速度，能更好的滿足業(yè)務(wù)方的需求。接下來逐一介紹。

第一代基于 Storm/Flink Java+Spark SQL

芒果 TV 的實時數(shù)據(jù)處理很早就開始了，最開始用的是 Storm，到了 18 年時，F(xiàn)link 橫空出世。Flink 的 State 與流處理的優(yōu)勢讓人眼前一亮，并且開源社區(qū)的大熱與大廠的相繼入坑，讓人無法拒絕，所以改用了 Flink 來搭建實時數(shù)倉，但當(dāng)時主要以滿足業(yè)務(wù)方需求為主，進行煙囪式的開發(fā)。基本流程是接上游 Kafka 的數(shù)據(jù)，使用 Flink Java 進行相關(guān)業(yè)務(wù)邏輯處理后，將數(shù)據(jù)輸出至對象存儲中。然后使用 Spark SQL 對數(shù)據(jù)進行統(tǒng)計等二次加工處理后，再交付客戶使用。此階段優(yōu)點是利用了 Flink 的長處，讓數(shù)據(jù)從源頭到終端更實時化了，滿足了業(yè)務(wù)方對數(shù)據(jù)的時效性與業(yè)務(wù)需求。缺點是來一個需求就做一個功能，并未有實時數(shù)倉的建設(shè)與沉淀。

第二代基于 Flink SQL+Spark SQL

基于上一階段的技術(shù)積累與發(fā)現(xiàn)的問題，提出了建設(shè)實時數(shù)倉的新方案。此時 Flink SQL 功能已初步完善，能滿足搭建數(shù)倉的各方面需求，SQL 化相較 Flink Java 也能降低開發(fā)、維護等各方面成本。于是選擇 Flink SQL 來搭建實時數(shù)倉。此階段對實時數(shù)倉進行了分層架構(gòu)設(shè)計，這個后面有詳細講解。基本流程是接上游 Kafka 數(shù)據(jù)進行格式化后輸出至 Kafka，下層接到 Kafka 數(shù)據(jù)進行字段處理、垃圾數(shù)據(jù)過濾等操作后輸出至 Kafka，最后一層接 Kafka 數(shù)據(jù)進行維度擴展，然后將數(shù)據(jù)寫至對象存儲中。再由 Spark SQL 讀取對象存儲中的數(shù)據(jù)進行統(tǒng)計等處理后，交付客戶使用。此階段的優(yōu)點是實現(xiàn)了數(shù)倉的分層構(gòu)架設(shè)計，對各層數(shù)據(jù)定義了標(biāo)準(zhǔn)化，實現(xiàn)了各層數(shù)據(jù)解耦，避免了煙囪式的開發(fā)，解決了重復(fù)開發(fā)等問題，實時數(shù)倉逐步走向成熟。缺點是使用 Spark SQL 進行后續(xù)統(tǒng)計與匯總時，不夠靈活。需要提前設(shè)計好指標(biāo)，面對客戶多變的需求時，往往不能很及時的響應(yīng)。

第三代基于 Flink SQL+StarRocks

隨著實時數(shù)倉的建設(shè)逐步加深，Spark SQL 不夠靈活，處理速度不夠快的弊端越發(fā)突出。此時 StarRocks 進入了我們的視線，其 MPP 的架構(gòu)、向量化引擎、多表 Join 等特性所展現(xiàn)出來在性能、易用性等方面的優(yōu)勢，都很好的彌補了 Spark SQL 在這塊的不足。于是經(jīng)調(diào)研后決定，在實時數(shù)倉中用 StarRocks 替換掉 Spark SQL 。在此階段，前面用 Flink SQL 搭建的實時數(shù)倉分層構(gòu)架并未改變，而下游用 Spark SQL 進行統(tǒng)計分析的相關(guān)功能，逐步替換成了用 StarRocks 來做。而基于 StarRocks 的優(yōu)勢與搭建實時數(shù)倉遇到的痛點，我們并沒有照搬之前 Spark SQL 的模式，而是選用了新的模式。使用 StarRocks 實現(xiàn)即席查詢。之前是用 Spark SQL 先將數(shù)據(jù)進行統(tǒng)計與匯總后，將最終結(jié)果數(shù)據(jù)寫入對象存儲中。而現(xiàn)在是直接用 StarRocks 對明細數(shù)據(jù)進行匯總，展示到前端頁面中。這么做的好處是能更快、更靈活的滿足業(yè)務(wù)方的需求，減少了開發(fā)工作量，減少了測試、上線等時間。StarRocks 優(yōu)秀的性能讓即席查詢速度并未變慢，功能更強大，更靈活，交付速度變更快了。

02

自研 Flink 實時計算調(diào)度平臺介紹

現(xiàn)有痛點

• 原生任務(wù)命令復(fù)雜，調(diào)試麻煩，開發(fā)成本比較高。
• 連接器， UDF，Jar 任務(wù)包等無法管理，調(diào)試復(fù)雜，經(jīng)常遇到依賴沖突問題。
• 無法做到統(tǒng)一的監(jiān)控報警以及對資源上的權(quán)限管理。
• SQL 任務(wù)開發(fā)復(fù)雜，沒有一個好用的編輯器和代碼管理及保存平臺。
• 基礎(chǔ)表， 維表， Catalog 沒有記錄和可視化的平臺。
• 多版本和跨云任務(wù)無法很好的管理。

• 沒有很好的日志管理機制，無法做到生產(chǎn)環(huán)境問題的快速定位。

平臺架構(gòu)設(shè)計

實時 Flink 調(diào)度平臺架構(gòu)圖：

平臺主要分為三個部分：

1. Phoenix Web 模塊主要負責(zé)面向用戶。

• 集群部署與任務(wù)提交。
• 公司各內(nèi)部業(yè)務(wù)權(quán)限管理。
• 支持 Catalog 及多源源信息管理。
• UDF，連接器等三方依賴 Jar 包管理。
• 多類型監(jiān)控報警以及日志管理。

• SQL 可視化編輯和校驗以及多版本存儲。

2. Flink SQL Gateway 和 Flink Jar Gateway 都是基于開源版本修改定制后的服務(wù)，支持 SQL 符合業(yè)務(wù)場景的解析和校驗以及 Jar 任務(wù)的提交，支持本地模式，Yarn-per-job 模式和 Application 模式，也支持自動的 Savepoint。

• 進行 SQL 的解析和校驗。
• 加載 SQL 和 Jar 任務(wù)所需要的三方依賴。
• SQL 任務(wù)連接 Catalog 存儲進行關(guān)聯(lián)和映射。
• Checkpoint 和 Savepoint 的自動管理和恢復(fù)。
• Jar 類型任務(wù)啟動參數(shù)的注入。
• 運行時配置的自適應(yīng)。

• 多類型的提交方式適配。

3. 混合多云模塊主要負責(zé)啟動任務(wù)的分發(fā)和云之間的信息管理。

03

 Flink SQL 實時數(shù)倉分層實踐

使用 Flink SQL 搭建實時數(shù)倉時，首要問題是數(shù)倉分層架構(gòu)如何解決，業(yè)界內(nèi)有許多優(yōu)秀的經(jīng)驗可以參考，同時也基于我們的情況，最終采用了如下數(shù)倉架構(gòu)：

ODS 層：原始日志層，在該層將上游 Binlog 日志、用戶行為日志、外部數(shù)據(jù)等數(shù)據(jù)源同步至數(shù)倉，對多種數(shù)據(jù)源、多種格式的數(shù)據(jù)通過統(tǒng)一 UDF 函數(shù)解析、格式化，最終輸出格式化 JSON 數(shù)據(jù)。

DW 層：數(shù)據(jù)明細層，在該層主要進行錯誤數(shù)據(jù)過濾、字段轉(zhuǎn)義、統(tǒng)一字段名等處理，輸出的數(shù)據(jù)已能滿足日常基礎(chǔ)分析的使用。

DM 層：數(shù)據(jù)模型層，在該層進行擴維，補充相關(guān)的公共信息。再按業(yè)務(wù)進行分域，輸出的數(shù)據(jù)具有更豐富的維度，可以滿足高級分析的數(shù)據(jù)使用需求。

ST 層：數(shù)據(jù)應(yīng)用層，按業(yè)務(wù)、功能等維度進行匯總，交由給前端頁面進行展現(xiàn)，輸出的數(shù)據(jù)可交付 Web、App、小程序等功能使用。

04

Flink SQL 實時數(shù)倉生產(chǎn)過程遇到的問題

在搭建實時數(shù)倉時，遇到了不少的問題，下面挑幾個典型的問題講解一下解決思路：

1. 多表關(guān)聯(lián)，這個在做數(shù)倉時，是個非常重要且普遍會要用到的功能，我們在使用 Flink SQL 搭建實時數(shù)倉初期，面對 Flink 琳瑯滿目的 Join 類型著實挑花了眼，尤其是涉及多表關(guān)聯(lián)時，有些維表的數(shù)據(jù)在 Hive 里，有些維表又在 MySQL 中，甚至還有些維表數(shù)據(jù)在其它 OLAP 中，該選擇何種關(guān)聯(lián)方式是當(dāng)時擺在面前的一大問題，后經(jīng)過多次嘗試，在性能、功能性等方面的多方權(quán)衡之下，總結(jié)出如下規(guī)則：

• 流表關(guān)聯(lián)維表（小數(shù)據(jù)量），使用 Lookup Join，維表數(shù)據(jù)量在十萬以下時，可使用 Hive 表做維表，因為離線數(shù)倉中的維表數(shù)據(jù)大部分都在 Hive 中，這樣的話就可以直接復(fù)用，省去數(shù)據(jù)導(dǎo)入導(dǎo)出的額外工作，并且性能方面沒有瓶頸，維表小時更新后，F(xiàn)link SQL 也能讀到最新數(shù)據(jù)。
• 流表關(guān)聯(lián)維表（大數(shù)據(jù)量），使用 Lookup Join，維表數(shù)據(jù)量在十萬 – 千萬以下時，可用 MySQL 表做維表，此時用 Hive 維表已不能滿足性能需求。可將數(shù)據(jù)導(dǎo)出至 MySQL 中，利用緩存機制，也能很好的滿足要求。

• 流表關(guān)聯(lián)流表，使用 Interval Join，通過兩個流表的時間字段來控制關(guān)聯(lián)范圍，這種關(guān)聯(lián)方式是目前用的比較多的。使用方式也跟離線比較接近。

2. 復(fù)雜的表處理，在一些數(shù)據(jù)清洗的復(fù)雜場景中，在關(guān)聯(lián)維表時，維表的數(shù)據(jù)會要經(jīng)過一層甚至多層的處理才能使用，離線數(shù)倉在這種場景下，可以直接在 Join 時寫多層子查詢來一步到位。但 Flink SQL 中不支持，在底層機制上就拒絕了。經(jīng)過多次嘗試與掙扎，最后采取的方案是在 Hive 中將維表數(shù)據(jù)進行預(yù)處理，實時數(shù)倉使用預(yù)處理后的維表數(shù)據(jù)。不過這只是一個過渡方案，目前從社區(qū)了解到，后續(xù)會有新的機制來實現(xiàn)在維表上進行任意的復(fù)雜計算后再做維表關(guān)聯(lián)。不得不說 Flink 社區(qū)的更新還是非常的快。

3. State 過大，在兩個流表進行關(guān)聯(lián)或進行匯總統(tǒng)計時，F(xiàn)link 的機制是會將數(shù)據(jù)緩存在 State 中。這就導(dǎo)致State 會過大，導(dǎo)致 GC 頻繁，進而任務(wù)失敗。針對這種情況，在研究了 Flink 的內(nèi)存機制后，得出的解決方案如下：

• 縮短時間范圍，根據(jù)業(yè)務(wù)需求，適當(dāng)減少關(guān)聯(lián)時兩條流的時間范圍。
• 調(diào)整 Managed Memory 大小，可以調(diào)整 Managed Memory 占比，適當(dāng)?shù)目s小其它內(nèi)存的使用。

• 設(shè)置 State 的 TTL 來避免緩存過多的數(shù)據(jù)。

4. 任務(wù)中頻繁出現(xiàn) Checkpoint expired before completing 異常，在實際生產(chǎn)環(huán)境中，發(fā)現(xiàn)有任務(wù)會頻繁的報這個錯誤，這個錯誤是說 Checkpoint 不能順利完成，因為 Flink 的 Checkpoint 有 Barrier 機制來保證數(shù)據(jù)的 ExactlyOnce 精確一致性，如果一批數(shù)據(jù)處理不完，Checkpoint 就完成不了，這塊有興趣的可以去了解一下。導(dǎo)致這個錯誤原因有多種，不同的問題也有不同的解答，接下來列舉一下各場景與解決方案：

• Checkpoint 超時時長太短，這個是比較常見，也比較好解決的一種情況。就是 Checkpoint 的超時時長設(shè)置的太短了，導(dǎo)致 Checkpoint 還沒完成就被報了超時，解決方案就是設(shè)置長一點，我們一般根據(jù)任務(wù)類型，會設(shè)置 6 秒-2 分鐘不等。
• 任務(wù)有背壓，這個也很常見，一個任務(wù)內(nèi)有多個操作，其中一個操作耗時較長影響了整個任務(wù)的執(zhí)行。也會影響 Checkpoint 的完成，這其中涉及到，有興趣的可以查一下。解決方案是可以從 WebUi 上找到執(zhí)行緩慢的 Task，具體問題具體分析，解決了就好了。
• 內(nèi)存不足，先說背景，我們在生產(chǎn)環(huán)境中一般使用 rocksdb statebackend，默認會保留全量 Checkpoint。而這種情況下，在遇到有關(guān)聯(lián)、分組統(tǒng)計等使用了 heap statebackend 的任務(wù)中，計算的中間結(jié)果會緩存到 State 中，State 的內(nèi)存默認是總內(nèi)存的 40%，在這種計算中會不太夠，從而導(dǎo)致頻率的 GC，也影響了 Checkpoint 的執(zhí)行。解決方案如下：
• 調(diào)大 TaskManager 的內(nèi)存，TaskManager 的內(nèi)存調(diào)大后，其它內(nèi)存區(qū)域也會相應(yīng)調(diào)大。
• 調(diào)大 Managed Memory 的內(nèi)存占比，就是設(shè)置 taskmanager.memory.managed.fraction 這個參數(shù)，可根據(jù)實際情況來，實際生產(chǎn)中最高可調(diào)到 90%。這種方法只調(diào)大了 ManagedMemory 一塊，如果內(nèi)存資源并不是很充裕時，可以用這種方式。

• 改用增量 Checkpoint，根據(jù)實際情況調(diào)整 State 的 TTL 時間，并開啟增量 Checkpoint。甚至都不用調(diào)內(nèi)存大小，也能解決問題。

5. 在 Flink SQL 中使用 if 函數(shù)時，一次偶然的發(fā)現(xiàn)，在返回 String 時，會按最大長度返回。什么意思呢，比如 if(condition， stringA， stringB)，stringA 的長度是 10，stringB 的長度是 2，如果 condition = false，返回 stringB 的時候，會補齊 stringB 的長度到 10，不夠的給空格。這是個需要注意的地方。但后續(xù)了解到目前該現(xiàn)象已在 1.16.3 版本修復(fù)了，而我們用的是 1.15，所以如果遇到了可以用 CaseWhen 替代或者升級 Flink 版本至 1.16.3 及以上即可解決。

05

StarRocks 選型背景及問題

 

在之前的的框架中我們是以Flink流式處理引擎完成原始日志的清洗，數(shù)據(jù)的打?qū)捙c輕度聚合，再落地到分布式文件系統(tǒng)或者對象存儲，通過離線 Spark SQL 五分鐘級別的調(diào)度批處理，結(jié)果會通過 Presto 等引擎去查詢，這樣的架構(gòu)在生產(chǎn)環(huán)境中漸漸顯露出很多問題。

例如：

• 存在重復(fù)計算的問題，原始數(shù)據(jù)會在不同的任務(wù)中反復(fù)清洗，有的需要多個原始數(shù)據(jù)的關(guān)聯(lián)也會反復(fù)的清洗，大量浪費了計算資源，代碼和數(shù)據(jù)流可重用性很差。
• 為了滿足離線批處理歷史累計值和當(dāng)前 5 分鐘窗口的計算指標(biāo)，在流量高峰期和當(dāng)日指標(biāo)累計到晚上時很可能在 5 分鐘之內(nèi)無法完成指標(biāo)的計算，有很大的超時風(fēng)險，業(yè)務(wù)會反饋實時指標(biāo)的延遲。
• 由于離線 Spark 批處理在多維組合分析并且又要求實時性情況下，略顯乏力。業(yè)務(wù)的在線化，催生出很多實時的場景，另一方面運營的精細化和分析的平民化也催生出多維的分析需求，這些場景下需要粒度特別細，維度特別豐富的底層數(shù)據(jù)，這兩部分的疊加起來就催生出了實時多維分析的場景。這時候我們需要不斷的增加維度組合，增加結(jié)果字段，增加計算資源來滿足以上場景，但是還是略顯乏力。
• 在數(shù)據(jù)時效性日益增加的今天，很多場景下數(shù)據(jù)的時效性提出了秒級毫秒級的要求，之前5分鐘級別的方式不能滿足業(yè)務(wù)需求。
• 在之前的實時任務(wù)中經(jīng)常需要在 Flink 內(nèi)存中做流和流的 Join，這些都需要在 Flink 任務(wù)內(nèi)存中做，由于上游多個數(shù)據(jù)流的數(shù)據(jù)到達時間不一致，很難設(shè)計合適的 window 去在計算引擎里打?qū)挃?shù)據(jù)，采用 Flink  Interval  Join 時多個流時間間隔太久狀態(tài)數(shù)據(jù)數(shù)據(jù)會非常龐大，啟用 mapState 之類的狀態(tài)計算又過于定制。
• 對于 Flink 清洗或者計算的結(jié)果可能需要多個存儲介質(zhì)中，對于明細數(shù)據(jù)我們可能會存儲在分布式文件系統(tǒng)或者對象存儲，這時候是 Flink+HDFS，對于業(yè)務(wù)更新流數(shù)據(jù)，可能是 Flink CDC+hbase(cassandra或者其他 key-value 數(shù)據(jù)庫)，對于 Flink 產(chǎn)生回撤流數(shù)據(jù)可能是 Flink+MySQL(redis)，對于風(fēng)控類數(shù)據(jù)或者傳統(tǒng)的精細化的看版可能是 Flink+ elasticsearch，對于大批量日志數(shù)據(jù)指標(biāo)分析可能是Flink+clickhouse，難以統(tǒng)一，資源大量損耗，維護成本同樣高。

• 在線上有大型活動或者大型節(jié)目時，實時數(shù)據(jù)量暴增，實時的大批量寫入的情況下，寫入延遲大，寫入效率不高，數(shù)據(jù)積壓。

總體分析，早期架構(gòu)有這樣一些問題。

• 數(shù)據(jù)源多樣，維護成本比較高。
• 性能不足，寫入延遲大，大促的場景會有數(shù)據(jù)積壓，交互式查詢體驗較差。
• 各個數(shù)據(jù)源割裂，無法關(guān)聯(lián)查詢，形成眾多數(shù)據(jù)孤島。然后從開發(fā)的角度，每個引擎都需要投入相應(yīng)的學(xué)習(xí)開發(fā)成本，程序復(fù)雜度比較高。
• 實時性要求高，并且開發(fā)效率快，代碼或者數(shù)據(jù)可重復(fù)利用性強。

• 實時任務(wù)開發(fā)沒有同一套標(biāo)準(zhǔn)，各自為戰(zhàn)。

為此我們在測試環(huán)境下做了簡單的性能對比，具體情況如下：

對比環(huán)境 StarRocks：4 *16C*128G Presto：22*32C*256G (非獨占)

數(shù)據(jù)量：事件表（共百億數(shù)據(jù)，日均千萬去重用數(shù)）

測試用例	Presto(s)	StarRocks(s)
單表聚合測試	13.1	5
關(guān)聯(lián)測試	19	8
留存	24	15
窗口函數(shù)	16	8
漏斗	3.5	3.2
多表關(guān)聯(lián)	36	19

本次測試使用了 4 臺16C128G 內(nèi)存的 BE 服務(wù)器，測試結(jié)論基本能夠滿足百億條數(shù)據(jù)的查詢需求。測試結(jié)果表明資源在相差很多的情況下，StarRocks 的性能還明顯優(yōu)于 Presto，且平均效率提升 2-3 倍。

06

基于 Flink SQL+StarRocks 實時分析數(shù)倉

基于已經(jīng)搭建完畢的 Flink SQL 的數(shù)倉分層體系，且由 StarRocks2.5X 版本升級到 StarRocks3.0X 存算分離版本并已大規(guī)模投入在生產(chǎn)環(huán)境中。

實時和離線湖倉一體的架構(gòu)圖：

明細模型

在大數(shù)據(jù)生產(chǎn)環(huán)境中最常見的日志數(shù)據(jù)，特點是數(shù)據(jù)量大，多維度靈活復(fù)雜的計算，計算指標(biāo)多，實時性強，秒級別的高性能查詢，簡單穩(wěn)定實時流寫入，大表的 Join，高基數(shù)去重。

這些要素對于  Flink SQL+StarRocks 都能滿足，首先實時平臺上使用 Flink SQL 快速對實時流日志數(shù)據(jù)進行清洗，打?qū)挘瑫r StarRocks 提供 Flink-Connector-StarRocks 連接器開箱即用，并且支持 ExactlyOnce 和事務(wù)支持，通過 Stream Load 低延時快速導(dǎo)入。

例如：

通過高效簡單的 Flink SQL 建表模式，批量百萬級寫入，速度快，同時生產(chǎn)環(huán)境單表十億級別以上數(shù)據(jù)計算多維度用戶訪問次數(shù)，和用戶去重數(shù)據(jù)，能達到秒級別。

主鍵模型

在 OLAP 數(shù)據(jù)倉庫中，可變數(shù)據(jù)通常是不受歡迎的。

對于數(shù)倉中的數(shù)據(jù)變更方式：

• 方式一：一些OLAP數(shù)據(jù)倉庫提供 Merge on Read 模型的更新功能，完成數(shù)據(jù)變更，例如(clickhouse)。

• 方式二：簡單來說就是創(chuàng)建新分區(qū)表，刪除老的分區(qū)表的數(shù)據(jù)，然后批量刷寫過去。

在新的分區(qū)中插入修改后的數(shù)據(jù)，通過分區(qū)交換完成數(shù)據(jù)變更。

通過批量刷寫的方式會要重新建表，刪除分區(qū)數(shù)據(jù)，刷寫數(shù)據(jù)過程繁雜，還可能導(dǎo)致出錯。

Merge on Read 模式在寫入時簡單高效，但讀取時會消耗大量的資源在版本合并上，同時由于 merge 算子的存在，使得謂詞無法下推、索引無法使用，嚴(yán)重的影響了查詢的性能。StarRocks 提供了基于 Delete and Insert 模式的主鍵模型，避免了因為版本合并導(dǎo)致的算子無法下推的問題。主鍵模型適合需要對數(shù)據(jù)進行實時更新的場景，可以更好的解決行級別的更新操作，支撐百萬級別的 TPS，特別適合 MySQL 或其他業(yè)務(wù)庫同步到 StarRocks 的場景。

而且通過 Flink CDC 和 StarRocks 完美結(jié)合可以實現(xiàn)業(yè)務(wù)庫到 OLAP 數(shù)據(jù)倉庫端到端的全量+增量的實時同步，一個任務(wù)可以搞定批量和實時的全部問題，并且高效穩(wěn)定。同時主鍵模型也可以解決 Flink 中回撤流輸出的問題，支持按條件更新，支持按列更新，這些都是傳統(tǒng) OLAP 數(shù)據(jù)庫很多不兼具的優(yōu)點。

Flink CDC+StarRocks 的模式可以解決生產(chǎn)環(huán)境中很多問題， StarRocks 和 Flink 結(jié)合去構(gòu)建實時數(shù)據(jù)分析體系的聯(lián)合解決方案，將在一定程度上顛覆既有的一些禁錮，形成實時數(shù)據(jù)分析新范式，加速融合實時日志數(shù)據(jù)和業(yè)務(wù)數(shù)據(jù)，也能解決傳統(tǒng)離線數(shù)據(jù)批量抽取的問題，實現(xiàn)了離線和實時在數(shù)據(jù)上的統(tǒng)一，加快流批一體的進程。

聚合模型

在實時數(shù)倉中還有一種場景，我們不太關(guān)心原始的明細數(shù)據(jù)，多為匯總類查詢，比如 SUM、MAX、MIN 等類型的查詢，舊數(shù)據(jù)更新不頻繁，只會追加新的數(shù)據(jù)，這個時候可以考慮使用聚合模型。建表時，支持定義排序鍵和指標(biāo)列，并為指標(biāo)列指定聚合函數(shù)。當(dāng)多條數(shù)據(jù)具有相同的排序鍵時，指標(biāo)列會進行聚合。在分析統(tǒng)計和匯總數(shù)據(jù)時，聚合模型能夠減少查詢時所需要處理的數(shù)據(jù)，提升查詢效率。

在之前我們可能會把這些操作放在 Flink 里面去統(tǒng)計，狀態(tài)數(shù)據(jù)會存在在內(nèi)存中，會導(dǎo)致狀態(tài)數(shù)據(jù)持續(xù)增長，并且消耗大量資源，將 Flink 的單純統(tǒng)計修改為 Flink SQL+StarRocks 聚合模型，F(xiàn)link 這里只需要對明細數(shù)據(jù)進行清洗并導(dǎo)入到 StarRocks，效率非常高且穩(wěn)定。

我們在實際生產(chǎn)中主要用來統(tǒng)計用戶觀看時長，點擊量，訂單統(tǒng)計等。

物化視圖

數(shù)據(jù)倉庫環(huán)境中的應(yīng)用程序經(jīng)常基于多個大表執(zhí)行復(fù)雜查詢，通常涉及多表之間數(shù)十億行數(shù)據(jù)的關(guān)聯(lián)和聚合。要實現(xiàn)這種實時多表關(guān)聯(lián)并查詢結(jié)果的方式，在之前我們可能會把此項內(nèi)容放在 Flink 實時數(shù)倉中去處理，分層處理關(guān)聯(lián)，合并，統(tǒng)計等任務(wù)，最后輸出結(jié)果層數(shù)據(jù)，處理此類查詢通常會大量消耗系統(tǒng)資源和時間，造成極高的查詢成本。

現(xiàn)在可以考慮使用 Flink SQL+StarRocks 的新思路去處理這種大規(guī)模的分層計算問題，使得 Flink SQL 這里只需要處理一些簡單清洗任務(wù)，把大量重復(fù)計算的邏輯下推到 StarRocks 去執(zhí)行，多個實時流實時落地，在 StarRocks 可以建立多級物化視圖的建模方式，StarRocks 的物化視圖不僅支持內(nèi)表和內(nèi)表關(guān)聯(lián)，也支持內(nèi)表和外表關(guān)聯(lián)，比如你的數(shù)據(jù)是在 MySQL，Hudi，Hive 等都可以通過 StarRocks 物化視圖的方式查詢加速，并設(shè)定定期刷新規(guī)則，從而避免手動調(diào)度關(guān)聯(lián)任務(wù)。其中最大的一個特點時，我們已經(jīng)建立的物化視圖，當(dāng)有新的查詢對已構(gòu)建了物化視圖的基表進行查詢時，系統(tǒng)自動判斷是否可以復(fù)用物化視圖中的預(yù)計算結(jié)果處理查詢。如果可以復(fù)用，系統(tǒng)會直接從相關(guān)的物化視圖讀取預(yù)計算結(jié)果，以避免重復(fù)計算消耗系統(tǒng)資源和時間。查詢的頻率越高或查詢語句越復(fù)雜，性能增益就會越很明顯。

實時即未來，StarRocks 在逐漸實現(xiàn)這樣的能力，StarRocks 和 Flink 結(jié)合去構(gòu)建實時數(shù)據(jù)分析體系的聯(lián)合解決方案，將在一定程度上顛覆既有的一些禁錮，形成實時數(shù)據(jù)分析新范式。

07

未來展望

湖倉一體

當(dāng)前芒果 TV 已經(jīng)實現(xiàn)了流批一體的數(shù)倉建設(shè)，而未來的重點是湖倉一體的建設(shè)。

數(shù)據(jù)湖的特點在于可以存儲各種類型和格式的原始數(shù)據(jù)，包括結(jié)構(gòu)化數(shù)據(jù)、半結(jié)構(gòu)化數(shù)據(jù)和非結(jié)構(gòu)化數(shù)據(jù)。而數(shù)據(jù)倉庫則是對數(shù)據(jù)進行結(jié)構(gòu)化和整理，以滿足特定的業(yè)務(wù)需求。

湖倉一體將數(shù)據(jù)倉庫和數(shù)據(jù)湖的特點融合在一起，打造一個統(tǒng)一的數(shù)據(jù)中心，實現(xiàn)對數(shù)據(jù)的集中管理。湖倉一體的架構(gòu)能夠提供更好的安全性、成本效益和開放性，既能夠存儲和管理大量原始數(shù)據(jù)，又能夠?qū)?shù)據(jù)整理成結(jié)構(gòu)化的形式，為分析和查詢提供便利。

通過建立湖倉一體，芒果 TV 能夠向公司內(nèi)部提供更豐富的數(shù)據(jù)服務(wù)，支持業(yè)務(wù)決策和創(chuàng)新，實現(xiàn)對數(shù)據(jù)的全面掌控和管理，包括數(shù)據(jù)的采集、存儲、處理和分析。同時，湖倉一體還能夠支持多種計算引擎和工具的使用，如 Flink、Spark、Hive 等，使得數(shù)據(jù)處理和分析更加靈活和高效。

低代碼

現(xiàn)在的開發(fā)方式是在自研的平臺上寫 SQL 提交任務(wù)，這種方式在面對一些清洗場景時，大部分是重復(fù)工作，有較大的提升空間。低代碼是時下比較熱門的概念，其在降本增效方面的優(yōu)勢很大。我們的下一步的計劃是逐步實現(xiàn)低代碼，第一階段是將實時平臺與數(shù)據(jù)上報平臺進行打通，通過讀取上報平臺里相關(guān)元數(shù)據(jù)，能夠自動生成對應(yīng)的數(shù)據(jù)清洗任務(wù)，解放生產(chǎn)力，提升工作效率與交付速度。

低代碼的優(yōu)勢在于它能夠?qū)㈤_發(fā)過程中的重復(fù)工作進行自動化和簡化，減少了開發(fā)人員的編碼工作量。通過可視化的方式，開發(fā)人員可以通過拖拽和配置來完成任務(wù)，而無需編寫大量的代碼。這不僅提高了開發(fā)效率，還降低了出錯的風(fēng)險。

通過實現(xiàn)低代碼的開發(fā)方式，芒果 TV 將能夠加快數(shù)據(jù)處理和分析的速度，提高團隊的整體效率。此外，低代碼還能夠降低對開發(fā)人員的技術(shù)要求，使得更多的人能夠參與到數(shù)據(jù)處理和分析的工作中。

總結(jié)而言，基于 Flink 技術(shù)的特點，芒果 TV 在未來的數(shù)倉建設(shè)中將注重實現(xiàn)湖倉一體的架構(gòu)，以實現(xiàn)對數(shù)據(jù)的全面管理和利用。同時，芒果 TV 計劃逐步實現(xiàn)低代碼的開發(fā)方式，以提高開發(fā)效率和交付速度。這些舉措將進一步推動芒果 TV 在長視頻數(shù)據(jù)分析領(lǐng)域的發(fā)展，為業(yè)務(wù)決策和創(chuàng)新提供更強大的支持。