本文整理自蔚來汽車大數(shù)據(jù)專家，架構(gòu)師吳江在 Flink Forward Asia 2021 行業(yè)實踐專場的演講。主要內(nèi)容包括：

• 實時計算在蔚來的發(fā)展歷程
• 實時計算平臺
• 實時看板
• CDP
• 實時數(shù)倉
• 其他應(yīng)用場景

一、 實時計算在蔚來的發(fā)展歷程

18 年 5 月份左右，我們開始接觸實時計算的概念，最初是用 Spark Streaming 做一些簡單的流式計算數(shù)據(jù)的處理;

19 年 9 月份我們引入了 Flink，通過命令行的方式進行提交，包括管理整個作業(yè)的生命周期;

到了 21 年 1 月份，我們上線了實時計算平臺 1.0，目前正在進行 2.0 版本的開發(fā)。

二、實時計算平臺

在實時計算平臺 1.0，我們是通過將代碼進行編譯，然后上傳 jar 包到一個服務(wù)器上，以命令行的方式進行提交。這個過程中存在很多問題：

• 首先，所有流程都是手動的，非常繁瑣而且容易出錯;
• 其次，缺乏監(jiān)控，F(xiàn)link 本身內(nèi)置了很多監(jiān)控，但是沒有一個自動的方式將它們加上去，還是需要手動地去做配置;
• 此外，任務(wù)的維護也非常麻煩，一些不太熟悉的開發(fā)人員進行操作很容易出現(xiàn)問題，而且出現(xiàn)問題之后也難以排查。

實時計算平臺 1.0 的生命周期如上圖。任務(wù)寫完之后打成 jar 包進行上傳提交，后續(xù)的開啟任務(wù)、停止、恢復(fù)和監(jiān)控都能夠自動進行。

作業(yè)管理主要負責作業(yè)的創(chuàng)建、運行、停止、恢復(fù)和更新。日志主要記錄 Flink 任務(wù)提交時的一些日志，如果是運行時的日志還是要通過 Yarn 集群里的 log 來查看，稍微有點麻煩。關(guān)于監(jiān)控和告警模塊，首先 metrics 監(jiān)控主要是利用 Flink 內(nèi)置的指標上傳到 Prometheus，然后配置各種監(jiān)控的界面，告警也是利用 Prometheus 的一些指標進行規(guī)則的設(shè)置，然后進行告警的設(shè)置。Yarn 負責整體集群資源的管理。

上圖是實時計算平臺 1.0 的界面，整體功能比較簡單。

上圖是實時計算平臺 2.0。相對于 1.0，最大的區(qū)別是藍色的部分。對于實時計算平臺的形態(tài)，可能并沒有一個統(tǒng)一的標準，它與每個公司本身的情況息息相關(guān)，比如公司本身的體量和規(guī)模、公司對實時計算平臺的資源投入等，最終還是應(yīng)該以適用于公司本身的現(xiàn)狀為最佳標準。

2.0 版本我們增加從開發(fā)到測試兩個階段功能的支持。簡單介紹一下它們的具體功能：

FlinkSQL：它是很多公司的實時計算平臺都支持的功能，它的優(yōu)點在于可以降低使用成本，也比較簡單易用。

空間管理：不同的部門和不同的組可以在自己的空間里進行作業(yè)的創(chuàng)建、管理。有了空間的概念之后，我們可以利用它做一些權(quán)限的控制，比如只能在自己有權(quán)限的空間里進行一些操作。

UDF 管理：使用了 FlinkSQL 的前提下，就可以基于 SQL 的語義用 UDF 的方式擴充功能。此外，UDF 還能用于 Java 和 Schema 任務(wù)，可以把一些公用的功能包裝成 UDF，降低開發(fā)成本。它還有一個很重要的功能就是調(diào)試，可以簡化原有的調(diào)試流程，做到用戶無感知。

實時計算平臺 2.0 的實現(xiàn)，帶給我們最大的影響就是減輕了數(shù)據(jù)團隊的負擔。在我們原先的開發(fā)流程里，經(jīng)常需要數(shù)據(jù)團隊的介入，但實際上其中的很大一部分工作都是比較簡單的，比如數(shù)據(jù)同步或數(shù)據(jù)的簡單處理，這類工作并不一定需要數(shù)據(jù)團隊去介入。

我們只需要把實時計算平臺做得更完善、易用和簡單，其他的團隊就可以使用 FlinkSQL 去做上述簡單的工作，理想的情況下他們甚至不需要知道 Flink 的相關(guān)概念就可以做一些 Flink 的開發(fā)。比如后臺人員做業(yè)務(wù)側(cè)開發(fā)的時候，對于一些比較簡單的場景就不需要依賴數(shù)據(jù)團隊，大大降低溝通成本，進度會更快。這樣在部門內(nèi)有一個閉環(huán)會更好一點。而且以這樣的方式，各個角色其實都會覺得比較開心。產(chǎn)品經(jīng)理的工作也會變得更輕松，在需求的階段不需要引入太多的團隊，工作量也會變少。

所以，這是一個以技術(shù)的方式來優(yōu)化組織流程的很好的例子。

三、實時看板

實時看板是一個比較常見的功能，在我們的具體實現(xiàn)中，主要發(fā)現(xiàn)了以下幾個難點：

第一，數(shù)據(jù)延遲上報。比如業(yè)務(wù)數(shù)據(jù)庫發(fā)生問題后，進行 CDC 接入的時候就需要中斷，包括后續(xù)寫到 Kafka，如果 Kafka 集群負載很高或 Kafka 發(fā)生問題，也會中斷一段時間，這些都會造成數(shù)據(jù)的延遲。上述延遲在理論上可以避免，但實際上很難完全避免。此外還有一些理論上就不能完全避免的延遲，比如用戶的流量或信號有問題導致操作日志無法實時上傳。

第二，流批一體。主要在于歷史數(shù)據(jù)和實時數(shù)據(jù)能否統(tǒng)一。

第三，維度的實時選擇。實時看板可能需要靈活選擇多個維度值，比如想先看北京的活躍用戶數(shù)，再看上海的活躍用戶數(shù)，最后看北京 + 上海的活躍用戶數(shù)，這個維度是根據(jù)需要可以靈活選擇的。

第四，指標的驗證。指標的驗證在離線的情況下，相對來說比較簡單一些，比如說可以做一些數(shù)據(jù)分布，看看每個分布的大概情況，也可以通過 ODS 層數(shù)據(jù)的計算與中間表進行比對，做交叉驗證。但是在實時的情況下就比較麻煩，因為實時處理是一直在進行的，有些情況很難去復(fù)現(xiàn)，此外也很難進行指標范圍或分布的驗證。

實時看板一般存在兩個方面的需求：

• 首先是時延方面，不同的場景對時延的要求是不同的，比如有些場景下能夠接受數(shù)據(jù)延遲 1-2 分鐘到達，但有的場景下只允許延遲幾秒鐘。不同場景下實踐的技術(shù)方案復(fù)雜度不一樣。
• 其次，需要兼顧實時與歷史看板的功能。有些場景下，除了需要看實時的數(shù)據(jù)變化，還需要對比著歷史數(shù)據(jù)來一起分析。

實時與歷史數(shù)據(jù)應(yīng)該進行統(tǒng)一的存儲，否則可能會存在很多問題。首先，實現(xiàn)的時候表結(jié)構(gòu)比較復(fù)雜，查詢的時候可能需要判斷哪段時間是歷史數(shù)據(jù)，哪段時間是實時數(shù)據(jù)，然后對它們進行拼接，會導致查詢的實現(xiàn)成本過高。其次，在歷史數(shù)據(jù)進行切換的時候也很容易出現(xiàn)問題，比如每天凌晨定時刷新歷史數(shù)據(jù)，此時如果歷史任務(wù)發(fā)生延遲或錯誤，很容易導致查出來的數(shù)據(jù)是錯誤的。

我們內(nèi)部對實時看板的延時性要求比較高，一般要求在秒級以內(nèi)，因為我們希望大屏幕上的數(shù)字是時刻在跳動和變化的。傳統(tǒng)的方案一般是采用拉的方式，比如說每秒查一次數(shù)據(jù)庫，實現(xiàn)的難度比較大，因為一個頁面會包含很多指標，需要同時發(fā)送很多接口去查詢數(shù)據(jù)，想讓所有數(shù)據(jù)都在一秒鐘之內(nèi)返回是不太可能的。另外，如果很多用戶同時進行查詢，會導致負載很高，時效性更難以保證。

所以我們采取了推的方式，上圖是具體實現(xiàn)的架構(gòu)圖，主要分為三層。第一層是數(shù)據(jù)層即 Kafka 的實時數(shù)倉，通過 Flink 對這些數(shù)據(jù)進行處理后將它們實時地推到后臺，后臺再實時地把它們推到前端。后臺與前端的交互是通過 web socket 來實現(xiàn)的，這樣就可以做到所有的數(shù)據(jù)都是實時推送。

在這個需求場景下，有一些功能會比較復(fù)雜。

舉個簡單例子，比如統(tǒng)計實時去重人數(shù) UV，其中一個維度是城市，一個用戶可能對應(yīng)多個城市，選擇上海和北京兩個城市的 UV 數(shù)，就意味著要把上海和北京的人放到一起進行去重，算出來去重的實時 UV 數(shù)據(jù)，這是一件比較麻煩的事情。從離線的角度來看，選多個維度是非常簡單的，把維度選好之后直接取出數(shù)據(jù)進行聚合即可。但是在實時場景下，要在哪些維度進行聚合是提前指定好的。

第一個方案是，在 Flink 狀態(tài)中存儲所有 user ID 和出現(xiàn)過的維度，并直接計算所有可能的維度組合 UV，然后將更新過的 UV 推送給前端。

但這種方式會增加很多計算成本，而且會導致維度爆炸，從而導致存儲成本也急劇增加。

第二種方案的架構(gòu)圖如上。我們把 sink 作為一個流式的核心，把端到端整體作為一個流式應(yīng)用，比如把數(shù)據(jù)的接入、在 Flink 中數(shù)據(jù)的處理計算、再到后臺、通過 web socket 推給前端這一整體作為一個應(yīng)用來考慮。

我們會在 Flink 里面存儲每個用戶所有的維度值，后臺的 Flink 推送的用戶具體情況也會存在每個城市下 user ID 的 list 里。Flink 擁有一個很關(guān)鍵的排除功能，如果用戶已經(jīng)出現(xiàn)過，那么在 Flink 階段就不會把變更推送到前端和后臺;如果用戶沒出現(xiàn)過，或者用戶出現(xiàn)過但城市沒出現(xiàn)過，那就會把用戶與城市的組合推送給后臺，保證后臺可以拿到每個城市下用戶 ID 去重的 list。

前端選擇維度之后，可以對后臺不同維度的 user ID 進行增量的訂閱。這里有兩個點需要注意：

第一是在前端剛打開在選擇緯度的時候，有一個初始化的過程，它會從后臺讀取所選維度的全量用戶 ID 來做一個合集，然后計算 UV 人數(shù)。

在第二個階段新的用戶 ID 到達之后，會通過 Flink 推送給后臺，而后臺只會推送增量 ID 給前端，然后前端因為已經(jīng)保存了之前的合集，對于增量的 ID，它就可以直接用 O(1) 的時間去算出新的合集，并且計算出它的 UV 人數(shù)。

可能有人會問，在這個方案下，用戶太多怎么辦?前端會不會占用太多的資源?

首先，從目前我們的實際使用場景來看，這個方案是夠用的，如果以后 ID 數(shù)激增，用 bitmap 也是一種選擇，但只用 bitmap 也不足以解決問題。因為不同公司用戶 ID 的生成規(guī)則不一樣，有些是自增 ID，有些是非自增 ID 或者甚至都不是一個數(shù)值，那就需要做映射，如果是一個離散的數(shù)值也需要額外做一些處理。

第一種方案把 ID 從 1 開始重新編碼，使它變得比較小且連續(xù)。目前大部分場景下大家可能都是用 RoaringBitMap，它的特點是如果 ID 非常稀疏，它在實際存儲的時候會使用一個 list 來存，而不是用 bitmap 來存，也就無法達到減少占用內(nèi)存的目的。所以要盡量讓 ID 的空間變小，讓 ID 的值比較連續(xù)。

但這樣還不夠，如果 ID 是之前沒出現(xiàn)過的，就需要給它重新分配一個 ID，但是處理這些數(shù)據(jù)的時候，F(xiàn)link task 的并行度可能大于 1，這個時候多個節(jié)點同時消費數(shù)據(jù)的話，它們可能都會遇到同樣的新 ID，如何給這個 ID 分配對應(yīng)的新的映射的小 ID?

舉個例子，一個節(jié)點查詢之后需要生成一個新 ID，同時又要保證其他節(jié)點不會再生成相同的 ID，可以通過在新 ID 上做唯一索引來保證，把索引創(chuàng)建成功就生成了新 ID，失敗的節(jié)點可以進行重試操作，去取現(xiàn)在的 ID mapping，因為剛才已經(jīng)有其他節(jié)點生成這個 ID 了，所以它在重試取 mapping 階段一定會成功。

除此之外，還需要考慮一種場景，比如用戶注冊完成后，馬上產(chǎn)生一些行為，而用戶注冊與一些業(yè)務(wù)模塊的行為表可能是由不同業(yè)務(wù)部門開發(fā)，也可能會存在不同的數(shù)據(jù)庫、不同的表里面，甚至是不同類型的數(shù)據(jù)庫，上述情況的接入方式也會不一樣，可能會導致雖然是先注冊，但是注冊數(shù)據(jù)流可能會稍微晚于行為數(shù)據(jù)流到達，這會不會導致出現(xiàn)什么問題?

目前看來是不會的，只需要行為數(shù)據(jù)流與新用戶注冊數(shù)據(jù)流共享一個 ID mapping 即可。

綜上，一個好的架構(gòu)，即使面對數(shù)據(jù)量激增的情況，也是不需要在架構(gòu)層面進行大改的，只需要在細節(jié)上進行重新設(shè)計。

第二個問題是前端會不會有很大的計算負載?

答案是：不會。雖然人數(shù)的去重是由前端來做，但只有前端第一次加載的時候才需要將用戶全量拉取，之后的增量 user ID 都會直接用 O(1) 的方式加入到目前的集合里，所以前端的計算負擔是很低的，整個過程完全是流式的。

第三個問題是實時報表同時訪問的用戶數(shù)很多怎么辦?

從目前的架構(gòu)上來看，對 Flink 和后臺側(cè)基本沒有影響，唯一的影響就是如果有很多用戶同時訪問，他們的頁面需要同時與后臺建立 web socket 連接。但是因為實時報表主要還是內(nèi)部使用，不會對外，所以同時的訪問量不會太多。

而且我們把數(shù)據(jù) ID 去重的一部分職責放在前端，即使有多個用戶同時訪問，計算職責也會分攤到不同的用戶瀏覽器里面去，實際上也不會有過多負載。

四、CDP

CDP 是一個運營平臺，負責偏后臺的工作。我們的 CDP 需要存儲一些數(shù)據(jù)，比如屬性的數(shù)據(jù)存在 ES 里、行為的明細數(shù)據(jù)包括統(tǒng)計數(shù)據(jù)存在 Doris 里、任務(wù)執(zhí)行情況存在 TiDB。也存在一些實時場景的應(yīng)用。

第一個是屬性需要實時更新，否則可能造成運營效果不佳。第二個是行為的聚合數(shù)據(jù)有時候也需要實時更新。

五、實時數(shù)倉

實時數(shù)倉重點考量點有以下幾個：

• 元信息管理，包括 Catalog 的管理。
• 分層，如何進行合理的分層。
• 建模，實時數(shù)倉應(yīng)該如何建模，它與離線數(shù)倉的建模方式有什么區(qū)別?
• 時效性，時延越低越好，鏈路越短越好。

上圖是我們目前的實時數(shù)倉架構(gòu)圖。它整體上與離線數(shù)倉非常相似，也是有一個原始層、DWD 層、DWS 層和 Application 層。

不同之處在于它有一個維度層 (DIM 層)，里面有很多不同的存儲介質(zhì)，維度信息可以放在 TiDB，并通過 AIO 的方式訪問維度表;也可以放在 Hive，用 Temporal Join 的方式去進行關(guān)聯(lián);有一些數(shù)據(jù)是一直在變化的，或者需要做一些基于時間的關(guān)聯(lián)，可以把數(shù)據(jù)放到 Kafka 里，然后用 Broadcast 或者 Temporal Join 去進行關(guān)聯(lián)。

左側(cè)是我們正在規(guī)劃中的能力。

第一個是血緣關(guān)系，它對于問題的溯源，以及對改動的影響的評估是有幫助的;

第二個是元信息管理，我們希望把所有數(shù)據(jù)都表化，在進行數(shù)據(jù)處理的時候可以直接用 SQL 搞定;

第三個是權(quán)限管理，對于不同的數(shù)據(jù)源、不同的表，都是需要做權(quán)限管理的;

第四個是數(shù)據(jù)質(zhì)量，如何進行數(shù)據(jù)質(zhì)量的保證。

下面是對這些未來規(guī)劃的具體闡述。

第一，Catalog 管理，這個功能目前暫未開發(fā)。我們希望為所有數(shù)據(jù)源創(chuàng)建一個表，不管里面的數(shù)據(jù)是維表還是其他表，是存在 MySQL 還是存在 Kafka，創(chuàng)建表之后都可以將這些細節(jié)屏蔽，通過 SQL 的方式就能輕松使用它。

第二，合理的分層。分層會對實時數(shù)倉造成多方面的影響。

首先，分層越多，時延越大。實時數(shù)倉是否需要這么多分層，值得深思。

其次，實時數(shù)據(jù)的質(zhì)量監(jiān)控會比離線數(shù)據(jù)更復(fù)雜，因為它是在不停地進行處理，分層越多，越難以發(fā)現(xiàn)問題、定位問題并進行回溯或復(fù)現(xiàn)，包括數(shù)據(jù)集成的分布也不易監(jiān)控。

最后，如何進行合理的分層?？隙ㄐ枰M可能減少層數(shù)，并且進行合理的業(yè)務(wù)功能垂直劃分，如果不同業(yè)務(wù)之間的交集很少，就盡量在各自業(yè)務(wù)領(lǐng)域內(nèi)建立自己單獨的分層。

第三，建模。這是離線數(shù)倉非常重要的部分，因為離線數(shù)倉非常大的一部分用戶是分析師，他們?nèi)粘９ぷ骶褪怯?SQL 進行數(shù)據(jù)的查詢和分析，這個時候就必須要考慮到易用性，比如大家都喜歡大寬表，所有相關(guān)字段都放到一個表里。所以在離線數(shù)倉建模和設(shè)計表結(jié)構(gòu)的時候，就需要盡量把一些可能用到的維度都加上。

而實時數(shù)倉面對的更多的是開發(fā)者，所以更強調(diào)實用性。因為在實時數(shù)倉的需求下，寬表里每增加一個字段都會增加時延，特別是維度的增加。所以說實時數(shù)倉的場景維表和建模更適合按實際需求來做。

第四，時效性。實時數(shù)倉本身還是需要有 raw 層，但是時效性比較高的場景，比如要同步一些線上的數(shù)據(jù)，這個數(shù)據(jù)最后同步快充也是線上的業(yè)務(wù)使用，要盡量減少鏈路，減少時延。比如可以用一些 Flink CDC 的方式減少中間層，這樣不單減少了整體的鏈路和時延，鏈路節(jié)點減少也意味著問題發(fā)生的概率變小。對于時延要求沒有那么高的內(nèi)部分析場景，盡量選擇使用實時數(shù)倉，可以減少數(shù)據(jù)的冗余。

六、其他應(yīng)用場景

其他的使用場景還包括 CQRS 類應(yīng)用。比如業(yè)務(wù)部門的功能更多的是考慮增刪改查或者是傳統(tǒng)數(shù)據(jù)庫操作，但后續(xù)還是會存在數(shù)據(jù)分析的場景，這個時候用業(yè)務(wù)庫去做分析是一個不太正確的方法，因為業(yè)務(wù)庫的設(shè)計本來就沒有考慮分析，更適合使用分析的 OLAP 引擎來做這項工作。這樣也就把業(yè)務(wù)部門要負責的工作和數(shù)據(jù)部門負責的工作分割開來，大家各司其職。

此外還有指標的監(jiān)控和異常檢測。比如對各種指標通過 Flink 進行實時的檢測，它會 load 一個機器學習模型，然后實時檢測指標的變化是否符合預(yù)期，和預(yù)期的差距有多大，還可以設(shè)置一個區(qū)域值來進行指標的異常檢測。

實時數(shù)據(jù)的場景越來越多，大家對實時數(shù)據(jù)的需求也越來越多，所以未來我們會繼續(xù)進行實時數(shù)據(jù)方面的探索。我們在流批一體的實時和離線存儲統(tǒng)一上已經(jīng)有了一些產(chǎn)出，我們也會在這方面投入更多精力，包括 Flink CDC 是否真的可以減少鏈路，提高響應(yīng)效率，也是我們會去考慮的問題。