本文整理自蔚來汽車大數據專家，架構師吳江在 Flink Forward Asia 2021 行業實踐專場的演講。主要內容包括：

• 實時計算在蔚來的發展歷程
• 實時計算平臺
• 實時看板
• CDP
• 實時數倉
• 其他應用場景

一、 實時計算在蔚來的發展歷程

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

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

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

二、實時計算平臺

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

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

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

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

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

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

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

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

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

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

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

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

所以，這是一個以技術的方式來優化組織流程的很好的例子。

三、實時看板

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

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

第二，流批一體。主要在于歷史數據和實時數據能否統一。

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

綜上，一個好的架構，即使面對數據量激增的情況，也是不需要在架構層面進行大改的，只需要在細節上進行重新設計。

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

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

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

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

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

四、CDP

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

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

五、實時數倉

實時數倉重點考量點有以下幾個：

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

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

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

左側是我們正在規劃中的能力。

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

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

第三個是權限管理，對于不同的數據源、不同的表，都是需要做權限管理的;

第四個是數據質量，如何進行數據質量的保證。

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

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

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

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

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

最后，如何進行合理的分層。肯定需要盡可能減少層數，并且進行合理的業務功能垂直劃分，如果不同業務之間的交集很少，就盡量在各自業務領域內建立自己單獨的分層。

第三，建模。這是離線數倉非常重要的部分，因為離線數倉非常大的一部分用戶是分析師，他們日常工作就是用 SQL 進行數據的查詢和分析，這個時候就必須要考慮到易用性，比如大家都喜歡大寬表，所有相關字段都放到一個表里。所以在離線數倉建模和設計表結構的時候，就需要盡量把一些可能用到的維度都加上。

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

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

六、其他應用場景

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

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

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