一、實時信號應用

騰訊內容中臺提供從內容生產、內容加工、內容分發、內容結算等全鏈路環節的一站式服務，在這個過程中，會產生大量的數據以及圍繞這些數據衍生的實時流業務應用，如內容周期智能管理，內容加工智能路由，內容創作精細運營等。

1、內容周期智能管理

為滿足不同用戶的體驗，需要給內容進行多種場景適配，隨著內容不斷增加，服務商成本非常高。為此，我們提供了一種基于內容周期提供分級服務的能力，在保障整體體驗的前提下，可有效降低成本。

2、內容加工智能路由

內容加工場景中有很多的加工精度，會抽象為一個微服務的編排系統。其中會有調度器控制任務的分發、路徑尋優、彈性伸縮等工作。通過實時的數據采集， 如算法、耗時、加工狀態等信息，并進行實時流加工，產生不同算法的實時效果特征信號，將這個實時信號反饋給調度器。可以進一步提高調度效果，減少加工耗時，提高處理成功率。

3、內容創作精細運營

互聯網平臺主要圍繞著技術、產品和運營，運營是一個非常關鍵的環節，運營人員會不斷發布精細化運營活動，創作者會領取運營任務并發文。基于實時計算的消費量、互動量等特征信號，可以進行活動達標判斷，進而將激勵實時觸達給創作者，提升運營活動效率。

二、問題與挑戰

在上述場景中會遇到很多問題和挑戰，可以抽象為四個方面：數據源、信號加工、信號觸發和數據治理。

1、數據源

騰訊內部各個業務方生產的數據各異，且擁有各自的 ID 體系；隨著業務發展，數據源還會動態添加消息 Topic，需要實時動態感知新增的數據源，并以中臺統一的 ID 視角串聯各個業務的內容數據。

2、信號加工

內容加工時會產生較多的復雜計算需求，比如，我們需要在有限資源內保障 TB 級多條實時數據拼接工作，以及長時間運行下需要對實時流應用的計算口徑進行調整而面臨的批數據重建流式數據狀態等問題，我們探索了一系列自研技術，解決了海量數據實時流計算問題。

3、信號觸發

內容生態系統很多場景依賴實時信號，并且基于規則進行控制和流轉，煙囪式開發有較大成本，我們需要構建一套日千億次匹配的規則引擎信號服務，保障資源共享，實現新增場景一鍵配置即可支持。

4、數據治理體系

我們希望建立全流程、全生命周期地建設數據治理體系。主要有以下幾個可通用的方法論：

（1）可觀測性：鏈路長，感知和定位問題慢。需要一個流程可觀測系統，幫助我們快速解決問題。

（2）退場機制：互聯網迭代非常快，探索性需求多，計算機系統資源開銷大。需要建立一套退場機制，當一些服務項失效的時候，及時將資源釋放，來降低服務的成本。

三、架構與解決方案

接下來，針對以上難點和挑戰，來介紹一下我們的解決方案。

1、整體架構

騰訊內容生態實時信號系統的整體架構如下圖所示。

自下而上來看，首先，原始數據包括各個業務渠道的生產流水、加工流水和消費流水等。 

接著是數據接入，通過動態多源、ID 映射、渠道衍生、數據清洗等能力，保證基礎數據的完整性和可用性。

再往上是信號生產，包括日千億次滑動大窗口計算，延遲流數據滾動大窗口計算，實時流數據拼接，融合批數據重建流狀態保障服務的不中斷，單體流量適應水平拓展保障程序不出現瓶頸，以及提供了輸出小文件數自適應流量的能力解決小文件過多的問題。 

規則引擎，主要提供統一的規則引擎和觸發系統，提供千億次的匹配，高效系統分發，以滿足各個業務系統的需求。 

數據治理，主要包括，保證系統的高可用性，一套可觀測性系統幫助我們快速地分析和解決問題，通過數據退場機制來降低成本，通過分層體系（常規的數據倉庫的建設，ODS 層，TWD 層，ADS 層等）保障數據的規范性和數據的可理解性，云數據管理系統將數據存在云端，供使用方查詢，并保障數據安全。

最上面是信號應用。 

2、數據接入——動態實時源自適應感知

騰訊內容中臺，提供一站式工業化的內容加工能力，每個業務方可自定義編排加工內容的任務流拓撲。為了穩定性和隔離性，每條任務流拓撲內容加工操作流水會生成一個 Topic，隨著業務發展，新的 Topic 會不斷增加，同時存量 Topic 數據量可能變大。因新增 Topic 所屬集群地址差異大，Flink Source 無法用正則匹配到，導致程序無法自動感知。因此，我們設計了 Topic 動態添加的自適應感知的技術方案，可以做到：

（1）數據完整性：自動感知新添加的拓撲 Topic，保證數據不遺漏。

（2）數據時效性：存量的 Topic 數據量級變大時，能夠自動擴容，保障整體時效性。

主要由以下幾個模塊構成：

（1）控制器模塊：監測消息隊列并通過配置中心異步控制Flink的消費。新增 Topic 時，注冊到配置中心。Topic 數據量變大導致消費延遲時，增加該 Topic 的消費并行度。

（2）配置中心：存放所有拓撲的消息隊列，如拓撲 ID、消費并行度、Kafka 配置。

（3）Flink 自適應 Source：自適應消費 Kafka 數據，保障數據完整性和時效性。在 Task 內開啟消費線程池，負責 Kafka 的消費；并有自適應 Client，負責控制線程池的消費，每分鐘執行一次，保障消費的完整性和時效性。具體步驟如下：

① 步驟 1：拉取所有消息隊列配置。

② 步驟 2：生成本 Task 消費的 Topic 消費列表，保障并行度 N 的 Topic 會被 N 個 Task 消費。總 Task 數目是 M，每個 Task 會被分到如下 Task 中：hash（pipeline_id）% M 到（hash（pipeline_id）+ N）% M。遍歷 Topic 可能被消費的 Task 列表，如果其中包含本 Task，則可對其進行消費。

③ 步驟 3：調整線程池消費列表，如果步驟 2 中添加了 Topic，則添加對應 Topic 的消費。

3、數據接入——十萬級 QPS 高并發 ID 映射

核心問題有兩點： 

（1）數據孤島：各個渠道有自己的內容 ID 體系。 

（2）資源開銷大：十萬級 QPS，IO 大，成本難以接受。

我們通過圖中所示的二級緩存，構建了一套高并發 ID 映射的解決方案，能夠打通中臺 ID，同時節省百倍的計算資源。

ID 映射有二級緩存，一級是在 Flink 內構建的渠道 ID 到中臺 ID 的映射，同時有一個遠程的第三方服務，可以提供實時的映射。整體的執行機制為，當有消費流水進來時，首先判斷在 Flink 應用內是否有緩存，如果有就直接返回中臺 ID，如果沒有，則進行遠程的 ID 映射，更新 Flink 狀態，返回中臺 ID，最后給業務輸出含中臺 ID 的消費流水。

我們在 Flink 中構建了兩種 cache，一種是可以映射的，將渠道 ID 映射到中臺 ID，另一種是未能映射的渠道 ID 和映射值。我們會采用不同的機制，保障數據的時效性和可用性。對于可以映射的情況，為了應對映射狀態的不斷膨脹，我們將 TTL 設置了 7 天，同時設置了 LRU 的緩存機制來進行控制。而未能映射的情況，可能是當時沒有映射上，而隔一段時間能夠通過第三方 ID 映射服務重新映射到中臺 ID。此時設置的 TTL 比較短，常為 1 小時。同時為了保障一段時間后仍然能映射上，采用了 FIFO 的機制，以保障映射的可用性，同時成本也能極大的降低。

4、信號生產——滑動大窗口計算

在內容場景中，需要對內容消費數據的大時間窗口（如 24 小時）的每分鐘滑動指標進行日千億次的實時流計算，并基于這樣的數據指標來控制業務流轉，如果我們直接基于 Flink 內部的窗口函數，進行實時計算窗口指標時，因不能及時關閉窗口，狀態數據會占用大量的內存，導致計算出現反壓的情況，程序穩定性差，容易出現卡死現象。

基于上述挑戰，我們設計了一種高性能的大窗口計算方法，主要有如下優點：

① 傳統的方式需要每次對大窗口中的全量數據做計算，而現有方式可以復用前一次計算結果，可極大減少計算量。

② 我們方案中大窗口是邏輯上的大窗口，相比 Flink 原生的窗口計算會保留大窗口時間內的原始數據，我們在內存中并不存放這些原始數據，只存放算法提到的聚合維度的數據，同時利用數據淘汰機制，防止內存占用過大，節約大量的內存資源。

對實時流數據根據數據自身的事件時間是否連續分為如下不同的幾種情況：

情況一：分鐘級別滑動，每分鐘窗口連續有流量的情況。

當數據自身的事件時間連續的時候，我們需要拿到上次大窗口的計算結果值，在上次計算結果的基礎上，對窗口的頭部和尾部進行加減操作就可以得到新的大窗口的值。

分鐘級滑動每分鐘連續的大窗口

其中，T（6, 4）代表的是 6min 時候近 4min 的累計值大小，其中 6 代表的是當前最新時間，4 代表的是需要統計的窗口大小，是人為規定的。M(5) 代表的是第 5min 的值。

情況二：分鐘級別滑動，每分鐘窗口流量不連續情況。

當間隔的時間小于窗口大小的情況下，計算當前窗口的值需要得到上一個窗口的值進行加減操作，由于數據自身的事件時間中斷，所以要對最后一次窗口的值進行校準。

分鐘級滑動每分鐘不連續大窗口

其中，T(5, 4) 代表的是 5min 時候近 4min 的累計值大小，其中 5 代表的是當前最新時間，4 代表的是需要統計的窗口大小，是人為規定的，M(5) 代表的是第 5min 的值。

情況三：分鐘級別滑動，每分鐘窗口流量不連續并且當間隔的時間大于窗口的情況。

當間隔的時間大于窗口大小的情況下，由于窗口時間內沒有出現流量，可以直接認為大窗口的計算值為當前分鐘流量值。

分鐘級滑動每分鐘不連續大窗口

其中，T（6, 4）代表的是 6min 時候近 4min 的累計值大小，其中 6 代表的是當前最新時間，4 代表的是需要統計的窗口大小，是人為規定的，M（5）代表的是第 5min 的值

5、信號生產--超大滑動窗口的計算

還有一種場景是超大滑動窗口的計算，每分鐘滑動一次，計算 30 天等超大滑動窗口指標。這種場景中狀態極大，資源開銷無法承受。以 30 天為例，如果只考慮半天的精度，可以將成本降低為千分之一，而精度只損失了百分之一，在成本和精度間達到了高效平衡。

如上圖所示，計算單個內容 ID 的超大滑動窗口指標過程如下。

（1）狀態更新：讀取消費流水，更新該 ID 的狀態值。

（2）計算超大窗口指標：基于應用內狀態進行計算。具體如下：

① 如果內容產生時間在 N 天內：取累計流量。

② 如果內容產生時間在 N 天前：基于輸入流量的時間取不同范圍的數據，整體半天精度，如 30 天超大窗口的誤差約 1.6%。時間區間劃分如下：

a）00:00—15:00：取過去 N 天+當天流量值。

b）15:00—23:59：取過去 N-1 天+當天流量值。

6、信號生產——TB 級實時流數據拼接

這里介紹的是 TB 級實時流數據拼接的場景。TB 級別數據拼接，計算慢、狀態易丟失，Flink 難以支持高可用。通過引入第三方 KV 存儲來備份狀態，保證了高可用和秒級時效。

主要思路如上圖所示，我們借助第三方 HBase 存儲完成多流關聯。

階段 1：特征拼接，每個源單獨加工，抽取自身特征后，進行如下過程：

① 步驟 1：將自身特征同步到 HBase 中，每個源只更新自身屬性對應的列。HBase 中會包含每個內容最新最全的屬性。

② 步驟 2：將有變更的內容推送到消息隊列中。當前實現是將所有有變更的內容實時推送下游，可改造該過程，多流水位對齊后再推送，以支持多流拼接的多種語義。

在本階段的存儲設計中，HBase 的 Rowkey 為待關聯的 Key，列分別為屬性 Key 和屬性值。同時，我們進行了大量優化設計：

① 批量訪問：每 50 個 Key 合并訪問，減少 IO。

② 隨機主鍵：將 Key 進行 md5 哈希，讓數據均勻分布在 HBase 中，防止熱點，提高隨機訪問性能。

③ 存儲壓縮：部分屬性值較大，將其序列化后，使用 GZIP 壓縮，減少存儲。

④ 過期機制：按需設置 TTL，防止數據無限膨脹。

階段 2：特征輸出，通過一個程序統一加工處理，可將每個內容的全量特征輸出到目標業務系統中。

① 步驟 3：實時感知特征有變更的內容。

② 步驟 4：批量拉取內容的全量特征，HBase 中每一列對應一個特征，某個內容的全部列即為其全部特征。

③ 步驟 5：入庫，將從 HBase 中獲取的全量特征，轉換成目標存儲格式，輸出到目標系統。

7、基于規則引擎的實時信號觸發器

根據很多配置規則，可以一鍵支持，秒級觸發，按照規則分發給業務系統。

首先在規則管理平臺，業務方配置規則；閾值可以通過預估模塊去訓練并進行更新，也可以手動設置。規則里面有靜態規則或者動態規則。同時有一些閾值服務，包括閾值同步和閾值查詢的能力。接著，在規則執行引擎，數據接入實時信號和消費流水，拉取到各個執行引擎里面，基于規則類型進行規則路由，分發到對應的動態規則匹配和固定規則模塊匹配，進行相應的信號觸發。配置加載，可以實時的加載閾值。設置信號去重的機制，可以保障同一信號短時間內不會重新觸發給下游。之后進行數據的分發，產生的信號會按照規則，分發到各自的系統。

規則類型主要分為固定規則和動態規則。固定規則，即所有內容的閾值相同；動態規則，不同內容閾值可以精細化設置。動態規則的人力成本較高，但是對一些成本非常高的場景，可以降低整體成本。

規則定義可以分為規則條件表達式和規則動作。比如騰訊視頻的流量大于多少就可以用一個條件表達式進行配置。同時會攜帶一些信息，比如去重周期等等。執行動作，是如何將匹配的信號分發給下游，通過 API 或者相應的消息隊列。

執行優化有三方面。

靈活引擎：基于 Flink + Aviator， 就可以構建一個分布式的，支持規則動態添加和修改的輕量級的規則匹配引擎。

二級緩存：針對每一個輸入信號，拉取其相關規則和閾值，因 IO 和 QPS 較高，整體成本非常大。二級緩存是規則 ID 和閾值 ID 直接去請求內容服務。然后引入二級緩存機制，通過拉取的閾值進行緩存之后，可以進行節省上百倍的資源。

預編譯：規則執行的過程為，首先將內容流量轉為 Map，表達式編譯為字節碼，進行執行，得到 True or False。如果是 False 就沒匹配上。在整個過程中消耗比較大的是將表達式編譯成字節碼。構建表達式到字節碼的緩存，該過程耗時會從 0.1ms 變成 0.04μs，在引入緩存的預編譯以后，單次規則匹配就可以節省上千倍的算力開銷。

8、數據治理——端到端全鏈路服務可觀測性

端到端的鏈路非常長，從 ODS 到 DWD， 到 DWS 層，到 ADS 層。對應一些延遲問題難以感知，有時需要業務方反饋才可知有延遲問題。同時當發現延遲以后的定位時間非常長。引入了服務可觀測系統的能力，將延遲的感知和定位問題環節從小時級縮短到分鐘級。

解決方案主要是引入了下面的幾個模塊：

數據染色：本模塊集成到各個加工程序中，將本環節和上游各個環節的時效性信息染色到輸出數據中，如事件時間、輸出時間等。

時效性統計：因為每個環節的輸出包含自身以及上游各個環節的時間信息，可基于某個環節的輸出數據，統計從數據源到當前環節端到端分環節、分數據源的時效性信息。

延遲監控：基于統計模塊計算的數據，監控端到端的延遲。

可觀測性分析工具：可以提供全鏈路的延遲分析。用戶可以選擇對應的主題，個性化設置延遲的閾值，分析不同節點的延遲情況，當有延遲時，可以快速定位延遲源頭。

9、數據治理——退場

因為探索性需求多，成本不斷膨脹，通過無效服務的下線，實現整體成本可控。通過看板無人使用的過程，與數據使用方的業務溝通來下線無人使用的數據看板。當確實無人使用的時候，把相對于的數據進行下線，節省成本。

同時還有一個解決方案是優化 TTL。部分服務場景起初需要去分析過去半年或者一年的數據，運行一段時間后，可能只用到過去一周的數據，這樣我們就可以根據訪問記錄去和用戶溝通。將一些熱數據保存在實時的 OLAP 系統里面，一些冷數據通過離線進行分析，降低我們資源的成本 。

10、數據治理——旁路系統保障狀態高可用

部分場景對數據的一致性要求非常高，但是開發過程中，依賴眾多，有小概率導致程序崩潰，當崩潰后程序狀態就丟失了。引入旁路系統后，可以保障核心狀態是 100% 可靠的。

我們構建了旁路系統，保障 Flink 狀態異常丟失后核心狀態的高可用。架構如圖所示，主要由兩個模塊構成

① 旁路系統：程序外起一個異步作業，將核心狀態從輸出中實時同步到 Redis 中。

② Flink 應用內狀態恢復模塊：為訪問 State 的前置邏輯，如果 Key 在應用內狀態中丟失，則從遠程 Redis 中恢復。

四、未來規劃

未來規劃主要在業務支撐、流批一體和資源優化三大方面。

（1）業務支撐。整體業務功能已經比較完備，接下來要更加關注精細化的運營需求，提高服務體驗。還要進一步實現核心能力的實時化，提高推薦效果和分析決策的效率。

（2）流批一體，實現一套代碼，兩種執行模式；一套系統，統一技術棧；一套運維，統一資源調度。數倉開發主要分為計算和存儲，存儲將使用數據湖模式，同時探索用 SQL 來統一離線和實時的技術棧，來保證更高效的開發。

（3）資源優化。順應降本增效的大環境，我們會探索資源彈性自適應技術的應用，和存儲優化，進一步降低成本。