一、背景介紹

1. 需要解決的業(yè)務痛點

推薦系統(tǒng)

對于推薦同學來說，想知道一個推薦策略在不同人群中的推薦效果是怎么樣的。

運營

對于運營的同學來說，想知道在廣東省的用戶中，最火的廣東地域內容是哪些？方便做地域 push。

審核

對于審核的同學，想知道過去 5 分鐘游戲類被舉報最多的內容和賬號是哪些，方便能夠及時處理。

內容創(chuàng)作

對于內容的作者，想知道今天到目前為止，內容被多少個用戶觀看，收到了多少個點贊和轉發(fā)，方便能夠及時調整他的策略。

老板決策

對于老板來說，想知道過去 10 分鐘有多少用戶消費了內容，對消費人群有一個宏觀的了解。

以上這幾點都是我們日常工作中經(jīng)常遇到的業(yè)務場景，后面的篇幅中會給出對應的解決方案。

2. 開發(fā)前調研

在進行開發(fā)之前我們做了如下這些調研。

■?2.1 離線數(shù)據(jù)分析平臺能否滿足這些需求

調研的結論是不能滿足離線數(shù)據(jù)分析平臺，不行的原因如下：

• 首先用戶的消費行為數(shù)據(jù)上報需要經(jīng)過 Spark 的多層離線計算，最終結果出庫到 MySQL 或者 ES 提供給離線分析平臺查詢。這個過程的延時至少是 3-6 個小時，目前比較常見的都是提供隔天的查詢，所以很多實時性要求高的業(yè)務場景都不能滿足。

• 另一個問題是騰訊看點的數(shù)據(jù)量太大，帶來的不穩(wěn)定性也比較大，經(jīng)常會有預料不到的延遲，所以離線分析平臺是無法滿足這些需求的。

■?2.2 準實時數(shù)據(jù)分析平臺

在騰訊內部提供了準實時數(shù)據(jù)查詢的功能，底層技術用的是 Kudu + Impala，Impala 雖然是 MPP 架構的大數(shù)據(jù)計算引擎，并且訪問以列式存儲數(shù)據(jù)的 Kudu。但是對于實時數(shù)據(jù)的分析場景來說，它的查詢響應速度和數(shù)據(jù)的延遲都還是比較高的。比如說查詢一次實時的 DAU 返回結果的耗時至少是幾分鐘，無法提供良好的交互式的用戶體驗。

所以 Kudu+Impala 這種通用的大數(shù)據(jù)處理框架的速度優(yōu)勢，更多的是相比 Spark 加 HDFS 這種離線分析框架來說的，對于我們實時性要求更高的場景是無法滿足的。因此需要進行開發(fā)，這就涉及到了方案選型和架構設計。

3. 騰訊看點信息流的業(yè)務流程

在大家介紹一下騰訊看點信息流的業(yè)務流程，了解了業(yè)務的流程，就能夠更好的理解技術架構的方案。

• 第 1 步，內容創(chuàng)作者發(fā)布內容；

• 第 2 步，內容會經(jīng)過內容審核系統(tǒng)啟用或者下架；

• 第 3 步，啟用的內容給到推薦系統(tǒng)和運營系統(tǒng)，分發(fā)給 C 側用戶；

• 第 4 步，內容分發(fā)給 C 側用戶之后，用戶會產(chǎn)生各種行為，比如說曝光、點擊舉報等，這些行為數(shù)據(jù)通過埋點上報，實時接入到消息隊列中；

• 第 5 步，構建實時數(shù)據(jù)倉庫；

• 第 6 步，構建實時數(shù)據(jù)查詢系統(tǒng)。

我們做的工作主要就在第 5 步和第 6 步，可以看一下我們的業(yè)務流程圖來進一步的了解。

在業(yè)務流程圖中，我們主要做的兩部分工作，就是圖中有顏色的這兩部分：

• 橙色部分，我們構建了一個騰訊看點的實時數(shù)據(jù)倉庫；

• 綠色部分，我們基于了 OLAP 的存儲計算引擎，開發(fā)了實時數(shù)據(jù)分析系統(tǒng)。

為什么要構建實時數(shù)據(jù)倉庫？因為原始的數(shù)據(jù)上報數(shù)據(jù)量非常大，一天上報的峰值就有上萬億條，而且上報的格式非常混亂，缺乏了內容的維度、信息用戶的畫像信息，下游就根本沒有辦法直接使用。

而我們提供的實時數(shù)據(jù)倉庫，是根據(jù)騰訊看點信息流的業(yè)務場景，進行了內容維度的關聯(lián)，用戶畫像的關聯(lián)和各種粒度的聚合，下游可以非常方便的使用實時數(shù)據(jù)，而且實時數(shù)據(jù)倉庫可以提供給下游的用戶反復的消費使用，可以大量的減少重復的工作。

綠色部分的多維實時數(shù)據(jù)分析系統(tǒng)，消費了我們提供的實時數(shù)據(jù)倉庫，利用了 OLAP 存儲計算引擎，將海量的數(shù)據(jù)進行高效的存儲，再提供高性能的多維實時分析功能。

二、架構設計

1. 設計的目標與難點

首先來看一下數(shù)據(jù)分析系統(tǒng)的設計目標與難點。我們的實時數(shù)據(jù)分析系統(tǒng)分為四大模塊：

• 實時計算引擎；

• 實時存儲引擎；

• 后臺服務層；

• 前端展示層。

難點主要在于前兩個模塊，實時計算引擎和實時存儲引擎。

• 千萬級每秒的海量數(shù)據(jù)如何實時的接入，并且進行極低延遲的維表關聯(lián)是有難度的；

• 實時存儲引擎如何支持高并發(fā)的寫入。高可用分布式和高性能的索引查詢是比較難的，可以看一下我們的系統(tǒng)架構設計來了解這幾個模塊的具體實現(xiàn)。

2. 系統(tǒng)架構設計

關于系統(tǒng)架構的設計，主要從以下幾方面來講。

■?2.1 實時計算

• 接入層主要是從千萬級每秒的原始消息隊列中拆分出不同業(yè)務不同行為數(shù)據(jù)的微隊列。拿 QQ 看點的視頻內容來說，拆分過后的數(shù)據(jù)就只有百萬級每秒了。

• 實時計算層主要是負責多行行為流水數(shù)據(jù)進行 "行轉列" 的操作，實時關聯(lián)用戶畫像數(shù)據(jù)和內容維度數(shù)據(jù)。

• 實時數(shù)倉存儲層主要就是設計出符合看點的業(yè)務，下游好用的實時消息隊列。

我們暫時提供了兩個消息隊列，作為實時數(shù)倉的兩層：

• 第一層是 DWM 層，它是內容 ID 和用戶 ID 粒度聚合的，就是說一條數(shù)據(jù)包含了內容 ID 和用戶 ID，然后還有 B 側的內容維度數(shù)據(jù)，C 側的用戶行為數(shù)據(jù)，還有用戶畫像數(shù)據(jù)。

• 第二層是 DWS 層，這一層是內容 ID 粒度聚合的，就是一條數(shù)據(jù)包含了內容 ID、B 側數(shù)據(jù)和 C 側數(shù)據(jù)?？梢钥吹絻热?ID 和用戶 ID 粒度的消息，隊列流量進一步減小到了 10 萬級每秒，內容 ID 粒度更是減小到了萬級每秒，并且格式更加清晰，維度信息更加豐富。

■?2.2 實時存儲

• 實時寫入層主要是負責 Hash 路由，將數(shù)據(jù)寫入；

• OLAP 存儲層是利用 MPP 的存儲引擎，設計出符合業(yè)務的索引和物化視圖，高效存儲海量數(shù)據(jù)；

• 后臺接口層是提供了高效的多維實時查詢接口。

■?2.3 后臺服務

后臺服務是基于騰訊自研的 RPC 后臺服務框架寫的，并且會進行一些二級緩存。

■?2.4 前端服務

前端采用的是開源組件 Ant Design，利用了 Nginx，反向代理了瀏覽器的請求到后臺服務器上。

3. 方案選型

關于架構設計的方案選型，我們對比了業(yè)內的領先方案，最終選擇了最符合我們業(yè)務場景的方案。

■ 3.1 實時數(shù)倉的選型

我們選擇的是業(yè)內比較成熟的 Lambda 架構，它的優(yōu)點是成熟度高，靈活性高，遷移成本低等等。但是它有一個缺點，實時和離線用了兩套代碼，可能會存在一個口徑修改了數(shù)據(jù)，但另一個沒有修改從而造成數(shù)據(jù)不一致的問題。我們的解決方案每天都有做數(shù)據(jù)對賬的工作，如果有異常會進行告警。

■?3.2 實時計算引擎的選型

我們選擇了 Flink 作為實時計算引擎，是因為 Flink 在設計之初就是為了流處理來設計的，Sparks Streaming 嚴格來說還是微批處理，storm 現(xiàn)在用的已經(jīng)不是很多了。并且， Flink 還有 exactly-once 的準確性，輕量級的容錯機制，低延遲高吞吐，應用性高的特點，所以我們選擇了 Flink 作為實時計算引擎。

■?3.3 實時存儲引擎

我們的要求是需要有維度索引，支持高并發(fā)的寫入和高性能的多維實時 OLAP 查詢。可以看到 HBase，TiDB 和 ES 都不能滿足要求。Druid 有一個缺陷，它是按照時序劃分 Segment，也就說明無法將同一個內容全部存放在同一個 Segment 上，所以在計算全局的 Top N 的時候就只能夠計算近似值。于是我們選擇了最近兩年大火的 MPP 數(shù)據(jù)庫引擎 Clickhouse，后面我會結合我們的具體使用場景和 Clickhouse 的內核原理，介紹一下 Clickhouse 的優(yōu)勢。

三、實時數(shù)倉

實時數(shù)倉也分為三塊來介紹：

• 第一是如何構建實時數(shù)倉；

• 第二是實時數(shù)倉的優(yōu)點；

• 第三是基于實時數(shù)倉，利用 Flink 開發(fā)實時應用時候遇到的一些問題。

實時數(shù)倉這一部分的難度在于它處于一個比較新的領域，并且各個公司各個業(yè)務的差距都比較大，怎么樣能夠設計出方便好用，符合看點信息流業(yè)務場景的實時數(shù)倉是有難度的。

1. 如何構建實時數(shù)倉

先看一下實時數(shù)倉要做什么。實時數(shù)倉對外來說就是幾個消息隊列，不同的消息隊列里面存放的是不同聚合粒度的實時數(shù)據(jù)，包括了內容 ID、用戶 ID、C 側用戶行為數(shù)據(jù)，B 側內容維度數(shù)據(jù)和用戶畫像數(shù)據(jù)等。搭建實時數(shù)倉可以分為三步。

■?1.1數(shù)據(jù)清洗

首先從海量的原始消息隊列中進行復雜的數(shù)據(jù)清洗操作，可以獲得格式清晰的實時數(shù)據(jù)。它的具體操作其實就是在 Flink 的實時計算環(huán)節(jié)，先按照一分鐘的粒度進行了窗口的聚合，在窗口內原本多行的行為數(shù)據(jù)被轉成了一行多列的數(shù)據(jù)格式。

■?1.2 高性能維表關聯(lián)

第二步是進行高性能的實時維表關聯(lián)，補充用戶畫像數(shù)據(jù)和內容維度數(shù)據(jù)等。但是海量的用戶畫像數(shù)據(jù)是存在于 HDFS 上的，內容維度數(shù)據(jù)又是存在于 HBase 上的，所以想要極低延遲的維表關聯(lián)是有技術挑戰(zhàn)的。這一塊在后文會單獨介紹。

■?1.3 不同粒度聚合

第三步是將算好的實時數(shù)據(jù)按照不同的粒度進行聚合，然后放到對應的消息隊列中進行保存，可以提供給下游多用戶復用，到這里實時數(shù)倉就搭建完成了。

接下來詳細介紹一下第二步中高性能實時維表關聯(lián)是怎么處理的。

幾十億的用戶畫像數(shù)據(jù)存放在 HDFS 上，肯定是無法進行高性能的維表關聯(lián)的，所以需要進行緩存。由于數(shù)據(jù)量太大，本地緩存的代價不合理，我們采用的是 Redis 進行緩存，具體實現(xiàn)是通過 Spark 批量讀取 HDFS 上的畫像數(shù)據(jù)，每天更新 Redis 緩存，內容維度數(shù)據(jù)存放在 HBase 中。

為了不影響線上的業(yè)務，我們訪問的是 HBase 的備庫，而且由于內容維度變化的頻率遠高于用戶畫像，所以維度關聯(lián)的時候，我們需要盡量的關聯(lián)到實時的 HBase 數(shù)據(jù)。

一分鐘窗口的數(shù)據(jù)，如果直接關聯(lián) HBase 的話，耗時是十幾分鐘，這樣會導致任務延遲。我們發(fā)現(xiàn) 1000 條數(shù)據(jù)訪問 HBase 是秒級的，而訪問 Redis 的話只是毫秒級的，訪問 Redis 的速度基本上是訪問 HBase 的 1000 倍，所以我們在訪問 HBase 的內容之前設置了一層 Redis 緩存，然后通過了監(jiān)聽 HBase-proxy 寫流水，通過這樣來保證緩存的一致性。

這樣一分鐘的窗口數(shù)據(jù)，原本關聯(lián)內容維度數(shù)據(jù)耗時需要十幾分鐘，現(xiàn)在就變成了秒級。我們?yōu)榱朔乐惯^期的數(shù)據(jù)浪費緩存，緩存的過期時間我們設置成了 24 個小時。

最后還有一些小的優(yōu)化，比如說內容數(shù)據(jù)上報過程中會上報不少非常規(guī)的內容 ID，這些內容 ID 在 HBase 中是不存儲的，會造成緩存穿透的問題。所以在實時計算的時候，我們直接過濾掉這些內容 ID，防止緩存穿透，又減少了一些時間。另外，因為設置了定時緩存，會引入一個緩存雪崩的問題，所以我們在實時計算的過程中進行了削峰填谷的操作，錯開了設置緩存的時間，來緩解緩存雪崩的問題。

2. 實時數(shù)倉的優(yōu)點

我們可以看一下，在我們建設實時數(shù)倉的前后，開發(fā)一個實時應用的區(qū)別。

沒有數(shù)倉的時候，我們需要消費千萬級每秒的原始隊列，進行復雜的數(shù)據(jù)清洗，然后再進行用戶畫像關聯(lián)、內容維度關聯(lián)，才能夠拿到符合要求格式的實時數(shù)據(jù)。開發(fā)和擴展的成本都會比較高。如果想開發(fā)一個新的應用，又要走一遍流程?，F(xiàn)在有了實時數(shù)倉之后，如果再想開發(fā)一個內容 ID 粒度的實時應用，就直接申請 TPS 萬級每秒的 DWS 層消息對列即可，開發(fā)成本變低很多，資源消耗小了很多，可擴展性也強了很多。

我們看一個實際的例子，開發(fā)我們系統(tǒng)的實時數(shù)據(jù)大屏，原本需要進行如上的所有操作才能夠拿到數(shù)據(jù)，現(xiàn)在只需要消費 DWS 層消息隊列寫一條 Flink SQL 即可，僅僅會消耗 2 個 CPU 核心和 1GB 的內存。以 50 個消費者為例，建立實時數(shù)倉的前后，下游開發(fā)一個實時應用，可以減少 98% 的資源消耗，包括了計算資源、存儲資源、人力成本和開發(fā)人員的學習接入成本等等，并且隨著消費者越多節(jié)省的就越多，就拿 Redis 存儲這一部分來說，一個月就能夠省下上百萬的人民幣。

3. Flink 開發(fā)過程中遇到的問題總結

在利用 Flink 開發(fā)實時應用的過程中遇到過不少問題，這里選擇幾個比較有代表性的給大家分享一下。

■?3.1 實時數(shù)據(jù)大屏

第一個是開發(fā)實時數(shù)據(jù)大屏的時候，開始是通過 Flink SQL 來實現(xiàn)的，功能非常簡單，就是計算當天截止到當前累計的點擊數(shù)，實現(xiàn)的方式也非常簡單，輸入的 source table 是實時數(shù)據(jù)倉庫的消息隊列。輸出的 sink table 就是 Redis。SQL 就是：select sum(click) from sourceTable Group by day time。

這個任務看起來是沒有問題的，但是實際跑起來數(shù)據(jù)卻無法實時更新，是因為 source table 每到達一條點擊數(shù)據(jù)，累計值都會加一，然后就會往 Redis 中寫一條最新的數(shù)據(jù)。所以當數(shù)據(jù)量太大的時候，它就會頻繁的寫 Redis，所以這樣就會導致寫 Redis 的網(wǎng)絡延遲會顯得非常高，從而會導致背壓數(shù)據(jù)無法實時更新。

我們做了一個簡單的優(yōu)化，用 table API 執(zhí)行完 SQL 之后，轉化成 DataStream，然后通過一個一秒鐘的數(shù)據(jù)窗口，每秒鐘僅僅會輸出最新的累計值到 Redis 中，這樣的數(shù)據(jù)就可以實時更新了。

■?3.2 Flink state 的 TTL

Flink 的 1.6 版本開始引入了 state TTL，開啟了 state TTL 之后，F(xiàn)link 就會為每一個 keyed state 增加一個時間戳字段，通過時間戳字段就可以判斷 state 是不是過期，是否需要進行清理。但是如果僅僅從字面意思上理解就會遇到一些問題，在 1.10 版本之前，雖然開啟了 state TTL，但是 Flink 默認是不會自動清理過期的 state 的。所以如果是 heap memory backend，就會導致 OOM 的問題；如果是 rocksDB backend，就會導致 state 的狀態(tài)越來越大，最終會導致重啟的時候耗費的時間過長。后面經(jīng)過調研，我們發(fā)現(xiàn)有兩種方式可以清理 Flink 的過期的 state。

第一種是手動清理，第二種的話是自動清理。我們最終選擇的是以手動觸發(fā)的方式來清理過期的 state。每天在深夜，也就是業(yè)務低谷期的時候，我們會對 state 中的數(shù)據(jù)進行遍歷的訪問，訪問到過期的數(shù)據(jù)，就會進行清理。

為什么我們沒有選擇 Flink 的自動清理策略，是因為 Flink 在 1.8 版本之前，只有一種自動清理策略，clean up in full snapshot。這種清理策略從名字上來看就知道他是在做全量 snapshot 的時候會進行清理，但是有一個致命的缺陷，它并不會減少本身 state 的大小，而是僅僅把清理過后的 state 做到 snapshot 里面，最終還是會 OOM。并且，它重啟之后才能夠加載到之前清理過的 state，會導致它頻繁的重啟。

雖然在 1.8 版本之后，增加了兩種自動清理的策略，但是因為它是異步清理，所以他的清理時機和使用方式都不如手動清理那么靈活，所以最終我們還是選擇了手動觸發(fā)的方式進行清理。在 1.10 版本之后，默認是選擇了自動清理的策略，但是這就要求用戶對自動清理策略的時機和策略 有比較好的了解，這樣才能夠更好的滿足業(yè)務的需求。

■?3.3 使用 Flink valueState 和 mapState 經(jīng)驗總結

雖然通過 valueState 也可以存儲 map 結構的數(shù)據(jù)，但是能夠使用 mapState 的地方盡量使用 mapState，最好不要通過 valueState 來存儲 map 結構的數(shù)據(jù)，因為 Flink 對 mapState 是進行了優(yōu)化的，效率會比 valuState 中存儲 map 結構的數(shù)據(jù)更加高效。

比如我們遇到過的一個問題就是使用 valueState 存儲了 map 結構的數(shù)據(jù)，選擇的是 rocksDB backend。我們發(fā)現(xiàn)磁盤的 IO 變得越來越高，延遲也相應的增加。后面發(fā)現(xiàn)是因為 valueState 中修改 map 中的任意一個 key 都會把整個 map 的數(shù)據(jù)給讀出來，然后再寫回去，這樣會導致 IO 過高。但是 mapState，它每一個 key 在 rocksDB 中都是一條單獨的 key，磁盤 IO 的代價就會小很多。

■?3.4 Checkpoint 超時問題

我們還遇到過一些問題，比如說 Checkpoint 超時了，當時我們第一個想法就是計算資源不足，并行度不夠導致的超時，所以我們直接增加了計算資源，增大了并行度，但是超時的情況并沒有得到緩解。后面經(jīng)過研究才發(fā)現(xiàn)是數(shù)據(jù)傾斜，導致某個節(jié)點的 barrier 下發(fā)不及時導致的，通過 rebalance 之后才能夠解決。

總的來說 Flink 功能還是很強的，它文檔比較完善，網(wǎng)上資料非常豐富，社區(qū)也很活躍，一般遇到問題都能夠比較快的找到解決方案。

四、實時數(shù)據(jù)查詢系統(tǒng)

我們的實時查詢系統(tǒng)，多維實時查詢系統(tǒng)用的是 Clickhouse 來實現(xiàn)的，這塊分為三個部分來介紹。第一是分布式高可用，第二是海量數(shù)據(jù)的寫入，第三是高性能的查詢。

Click house 有很多表引擎，表引擎決定了數(shù)據(jù)以什么方式存儲，以什么方式加載，以及數(shù)據(jù)表擁有什么樣的特性？目前 Clickhouse 擁有 merge tree、replaceingMerge Tree、AggregatingMergeTree、外存、內存、IO 等 20 多種表引擎，其中最體現(xiàn) Clickhouse 性能特點的是 merge tree 及其家族表引擎，并且當前 Clickhouse 也只有 merge 及其家族表引擎支持了主鍵索引、數(shù)據(jù)分區(qū)、數(shù)據(jù)副本等優(yōu)秀的特性。我們當前使用的也是 Clickhouse 的 merge tree 及其家族表引擎，接下來的介紹都是基于引擎展開的。

1. 分布式高可用

先看分布式高可用，不管單節(jié)點的性能多強，隨著業(yè)務的增長，早晚都會有遇到瓶頸的一天，而且意外的宕機在計算機的運行中是無法避免的。Clickhouse 通過分片來水平擴展集群，將總的數(shù)據(jù)水平分成 m 分，然后每個分片中保存一份數(shù)據(jù)，避開了單節(jié)點的性能瓶頸，然后通過副本即每個分片擁有若干個數(shù)據(jù)一樣的副本來保障集群的高可用。

再看看 Clickhouse 默認的高可用方案，數(shù)據(jù)寫入是通過分布式表寫入，然后分布式表會將數(shù)據(jù)同時寫入到同一個分片的所有副本里面。這里會有一個問題，如果副本 0 寫入成功，副本 1 寫入失敗，那么就會造成同一個分片的不同副本數(shù)據(jù)不一致的問題，所以默認的高可用方案是不能夠用于生產(chǎn)環(huán)境的。

我們這里聽取的是 Clickhouse 官方的建議，借助了 Zookeeper 實現(xiàn)高可用的方案，數(shù)據(jù)寫入一個分片的時候，僅僅寫入一個副本，然后再寫 Zookeeper，通過 Zookeeper 告訴同一個分片的其他副本，再過來拉取數(shù)據(jù)，保證數(shù)據(jù)的一致性。

接下來看一下 Clickhouse 實現(xiàn)這種高可用方案的底層原理，這種高可用的方案需要通過 Clickhouse 的 replicated merge tree 表引擎來實現(xiàn)，其中在 replicated merge tree 表引擎的核心代碼中，有大量跟 Zookeeper 進行交互的邏輯，從而實現(xiàn)了多個副本的協(xié)同，包括主副本的選舉寫入任務隊列的變更和副本狀態(tài)的變化等等?？梢钥吹酵獠繑?shù)據(jù)寫入 Clickhouse 的一個分片，會先寫入一個副本的內存中，在內存中按照指定的條件排好序，再寫入磁盤的一個臨時目錄。最后將臨時目錄重命名為最終目錄的名字，寫完之后通過 Zookeeper 進行一系列的交互，實現(xiàn)數(shù)據(jù)的復制。

這里沒有選用消息隊列進行數(shù)據(jù)的同步，是因為 Zookeeper 更加輕量級，而且寫的時候任意寫一個副本，其他的副本都能夠通過讀 Zookeeper 獲得一致性的數(shù)據(jù)，而且就算其他節(jié)點第一次來獲取數(shù)據(jù)失敗了，后面只要發(fā)現(xiàn)它跟 Zookeeper 上的數(shù)據(jù)記錄不一致，就會再次嘗試獲取數(shù)據(jù)，保證數(shù)據(jù)的一致性。

2. 海量數(shù)據(jù)的寫入

■?2.1 Append + Merge

數(shù)據(jù)寫入遇到的第一個問題是海量數(shù)據(jù)直接寫 Clickhouse 是會失敗的。Clickhouse 的 merge tree 家族表引擎的底層原理類似于 LSM tree，數(shù)據(jù)是通過 append 的方式寫入，后續(xù)再啟動 merge 線程，將小的數(shù)據(jù)文件進行合并。了解了 Clickhouse merge tree 家族表引擎的寫入過程，我們就會發(fā)現(xiàn)以下兩個問題。

• 如果一次寫入的數(shù)據(jù)太少，比如一條數(shù)據(jù)只寫一次，就會產(chǎn)生大量的文件目錄。當后臺合并線程來不及合并的時候，文件目錄的數(shù)量就會越來越多，這會導致 Clickhouse 拋出 too many parts 的異常，寫入失敗。

• 另外，之前介紹的每一次寫入除了數(shù)據(jù)本身，Clickhouse 還會需要跟 Zookeeper 進行 10 來次的數(shù)據(jù)交互，而我們知道 Zookeeper 本身是不能夠承受很高的并發(fā)的，所以可以看到寫入 Clickhouse QPS 過高，導致 zookeeper 的崩潰。

我們采用的解決方案是改用 batch 的方式寫入，寫入 zookeeper 一個 batch 的數(shù)據(jù)，產(chǎn)生一個數(shù)據(jù)目錄，然后再與 Zookeeper 進行一次數(shù)據(jù)交互。那么 batch 設置多大？如果 batch 太小的話，就緩解不了 Zookeeper 的壓力；但是 batch 也不能設置的太大，要不然上游的內存壓力以及數(shù)據(jù)的延遲都會比較大。所以通過實驗，最終我們選擇了大小幾十萬的 batch，這樣可以避免了 QPS 太高帶來的問題。

其實當前的方案還是有優(yōu)化空間的，比如說 Zookeeper 無法線性擴展，我有了解到業(yè)內有些團隊就把 Mark 和 date part 相關的信息不寫入 Zookeeper。這樣能夠減少 Zookeeper 的壓力。不過這樣涉及到了對源代碼的修改，對于一般的業(yè)務團隊來說，實現(xiàn)的成本就會比較高。

■?2.2 分布式表寫入

如果數(shù)據(jù)寫入通過分布式表寫入會遇到單點的磁盤問題，先介紹一下分布式表，分布式表實際上是一張邏輯表，它本身并不存儲真實的數(shù)據(jù)，可以理解為一張代理表，比如用戶查詢分布式表，分布式表會將查詢請求下發(fā)到每一個分片的本地表上進行查詢，然后再收集每一個本地表的查詢結果，匯總之后再返回給用戶。那么用戶寫入分布式表的場景，是用戶將一個大的 batch 的數(shù)據(jù)寫入分布式表，然后分布式表示按照一定的規(guī)則，將大的 batch 的數(shù)據(jù)劃分為若干個 mini batch 的數(shù)據(jù)，存儲到不同的分片上。

這里有一個很容易誤解的地方，我們最開始也是以為分布式表只是按照一定的規(guī)則做一個網(wǎng)絡的轉發(fā)，以為萬兆網(wǎng)卡的帶寬就足夠，不會出現(xiàn)單點的性能瓶頸。但是實際上 Clickhouse 是這樣做的，我們看一個例子，有三個分片 shard1，shard2 和 shard3，其中分布式表建立在 shard2 的節(jié)點上。

• 第一步，我們給分布式表寫入 300 條數(shù)據(jù)，分布式表會根據(jù)路由規(guī)則把數(shù)據(jù)進行分組，假設 shard1 分到 50 條，shard2 分到 150 條，shard3 分到 100 條。

• 第二步，因為分布式表跟 shard2 是在同一臺機器上，所以 shard2 的 150 條就直接寫入磁盤了。然后 shard1 的 50 條和 shard3 的 100 條，并不是直接轉發(fā)給他們的，而是也會在分布式表的機器上先寫入磁盤的臨時目錄。

• 第三步，分布式表節(jié)點 shard2 會向 shard1 節(jié)點和 shard3 節(jié)點分別發(fā)起遠程連接的請求，將對應臨時目錄的數(shù)據(jù)發(fā)送給 shard1 和 shard3。

這里可以看到分布式表所在的節(jié)點 shard2 全量數(shù)據(jù)都會先落在磁盤上，我們知道磁盤的讀寫速度都是不夠快的，很容易就會出現(xiàn)單點的磁盤性能瓶頸。比如單 QQ 看點的視頻內容，每天可能寫入百億級的數(shù)據(jù)，如果寫一張分布式表，很容易就會造成單臺機器出現(xiàn)磁盤的瓶頸，尤其是 Clickhouse 的底層運用的是 merge tree，它在合并的過程中會存在寫放大的問題，這樣會加重磁盤的壓力。

我們做的兩個優(yōu)化方案:

• 第一個就是對磁盤做了 RAID 提升了磁盤的 IO；

• 第二就是在寫入之前，上游進行了數(shù)據(jù)的劃分分表操作，直接分開寫入到不同的分片上，磁盤的壓力直接變?yōu)榱嗽瓉淼?n 分之一，這樣就很好的避免了磁盤的單點的瓶頸。

■?2.3 局部 Top 并非全局 Top

雖然我們的寫入是按照分片進行了劃分，但是這里引入了一個分布式系統(tǒng)常見的問題，就是局部的 Top 并非全局 Top。比如說同一個內容 x 的數(shù)據(jù)落在了不同的分片上，計算全局 Top100 點擊內容的時候，之前說到分布式表會將查詢請求下發(fā)到各個分片上，計算局部的 Top100 點擊的內容，然后將結果進行匯總。

舉個例子，內容 x 在分片一和分片二上不是 Top100，所以在匯總數(shù)據(jù)的時候就會丟失掉分片一和分片二上的內容 x 的點擊數(shù)據(jù)。

第二是會造成數(shù)據(jù)錯誤，我們做的優(yōu)化就是在寫入之前加上了一層路由，我們將同一個內容 ID 的數(shù)據(jù)全部路由到了同一個分片上，解決了該問題。這里需要多說一下，現(xiàn)在最新版的 Clickhouse 都是不存在這樣這個問題的，對于有 group by 和 limit 的 SQL 命令，只把 group by 語句下發(fā)到本地表進行執(zhí)行，然后各個本地表執(zhí)行完的全量結果都會傳到分布式表，在分布式表再進行一次全局的 group by，最后再做 limit 的操作。

這樣雖然能夠保證全局 top N 的正確性，但代價就是犧牲了一部分的執(zhí)行性能。如果想要恢復到更高的執(zhí)行性能，我們可以通過 Clickhouse 提供的 distributed_group_by_no_merge 參數(shù)來選擇執(zhí)行的方式。然后再將同一個內容 ID 的記錄全部路由到同一個分片上，這樣在本地表也能夠執(zhí)行 limit 操作。

3. 高性能的存儲和查詢

Clickhouse 高性能查詢的一個關鍵點，就是稀疏索引。稀疏索引這個設計很有講究，設計的好可以加速查詢，設計的不好反而會影響查詢效率。因為我們的查詢大部分都是時間和內容 ID 相關的，比如說某個內容過去 n 分鐘在各個人群的表現(xiàn)如何，我按照日期分鐘粒度時間和內容 ID 建立了稀疏索引，針對某個內容的查詢，建立稀疏索引之后，可以減少 99% 的文件掃描。

Clickhouse 高性能查詢的第二點，就是我們現(xiàn)在的數(shù)據(jù)量太大，維度太多，拿 QQ 看點的視頻內容來說，一天入庫的流水就有上百億條，有些維度有幾百個類別，如果一次性把所有的維度進行預聚合查詢反而會變慢，并且索引會占用大量的存儲空間。我們的優(yōu)化就是針對不同的維度建立對應的預聚合和物化視圖，用空間換時間，這樣可以縮短查詢的時間。

舉個例子，通過 summary merge tree 建立一個內容 ID 粒度聚合的累積，累加 pv 的物化視圖，這樣相當于提前進行了 group by 的計算，等真正需要查詢聚合結果的時候，就直接查詢物化視圖，數(shù)據(jù)都是已經(jīng)聚合計算過的，且數(shù)據(jù)的掃描量只是原始流水的千分之一。

分布式表查詢還會有一個問題，就是查詢單個內容 ID 的時候，分布式表會將查詢請求下發(fā)到所有的分片上，然后再返回給查詢結果進行匯總。實際上因為做過路由，一個內容 ID 只存在于一個分片上，剩下的分片其實都是在空跑。針對這類的查詢，我們的優(yōu)化就是后臺按照同樣的規(guī)則先進行路由，然后再查詢目標分片，這樣減少了 n 分之 n -1 的負載，可以大量的縮短查詢時間。而且由于我們提供的是 OLAP 的查詢，數(shù)據(jù)滿足最終的一致性即可。所以通過主從副本的讀寫分離，也可以進一步的提升性能。我們在后臺還做了一個一分鐘的數(shù)據(jù)緩存，這樣針對相同條件的查詢，后臺就可以直接返回。

4. Clickhouse 擴容方案

我們調研了業(yè)內一些常見的方案：

• 比如說 HBase 原始數(shù)據(jù)是存放在 HDFS 上的，擴容只是 region server 的擴容，并不涉及到原始數(shù)據(jù)的遷移。

• 但是 Clickhouse 的每個分片數(shù)據(jù)都是在本地，更像是 RocksDB 的底層存儲引擎，不能像 HBase 那樣方便的擴容。

• 然后是 Redis，Redis 是 Hash 槽這一種，類似于一致性 Hash 的方式，是比較經(jīng)典的分布式緩存方案。

Redis slot 在 Hash 的過程中，雖然會存在短暫的 ASK 不可用，但是總體上來說遷移還是比較方便的。就從原來的 h0 遷移遷移到 h1，最后再刪除 h0，但是 Clickhouse 大部分都是 OLAP 的批量查詢，而且由于列式存儲不支持刪除的特性，一致性 hash 的方案也不是很合適。

我們目前的擴容方案就是從實時數(shù)倉另外消費一份數(shù)據(jù)寫入新的 Clickhouse 集群，兩個集群一起跑一段時間，因為實時數(shù)據(jù)我們現(xiàn)在就保存了三天，等三天之后，后臺服務就直接訪問新的 Clickhouse 集群。

五、實時系統(tǒng)應用成果總結

我們輸出了騰訊看點的實時數(shù)據(jù)倉庫，DWM 層和 DWS 層兩個消息隊列，上線了騰訊看點的實時數(shù)據(jù)分析系統(tǒng)，該系統(tǒng)能夠亞秒級的響應多維條件查詢請求。在未命中緩存的情況下：

• 過去 30 分鐘的內容查詢，99% 的請求耗時在一秒內；

• 過去 24 小時的內容查詢 90% 的請求耗時在 5 秒內，99% 的請求耗時在 10 秒內。