JUST技術：CK實現時序數據管理

本次JUST系列技術分享，京東數科為您帶來是，JUST（https://just.urban-computing.cn/）是如何使用ClickHouse實現時序數據管理和挖掘的。ClickHouse是一個高效的開源聯機分析列式資料庫管理系統，由俄羅斯IT公司Yandex開發的，並於2016年6月宣布開源。

一、時序數據簡介

時序數據全稱是時間序列（TimeSeries）數據，是按照時間順序索引的一系列數據點。最常見的是在連續的等時間間隔時間點上獲取的序列，因此，它是一系列離散數據[1]。

時序數據幾乎無處不在，在目前單向的時間流中，人的脈搏、空氣的溼度、股票的價格等都隨著時間的流逝不斷變化。時序數據是數據的一種，因為它顯著而有價值的特點，成為我們特別分析的對象。

將時序數據可以建模為如下部分組成：

• Metric：度量的數據集，類似於關係型資料庫中的 table，是固定屬性，一般不隨時間而變化
• Timestamp：時間戳，表徵採集到數據的時間點
• Tags：維度列，用於描述Metric，代表數據的歸屬、屬性，表明是哪個設備/模塊產生的，一般不隨著時間變化
• Field/Value：指標列，代表數據的測量值，可以是單值也可以是多值

一個具體的多值模型時序數據案例如表1所示：

表1 時序數據案例

二、時序數據管理概述

2.1 時序時序管理的流程

一切數據的本質都是為價值服務的，獲取價值的這個過程就是數據管理與分析。從技術上來說，任何數據從產生到滅亡都會經歷如圖1所示的過程。

圖1 數據生命周期

時序數據也不例外，只是每個部分的處理不同。

（1）數據採集。同一個場景下時序數據產生的頻率一般恆定，但在不同場景下採集數據的頻率是變化的，每秒一千條和每秒一千萬條數據使用的技術是完全不同的。所以，數據採集要考慮的主要是頻率和並發。

（2）數據存儲。數據存儲是為了查詢和分析服務的。以什麼格式存儲、建什麼索引、存儲數據量大小、存儲時長是時序數據存儲要考慮的，一般時序數據寫多讀少，數據具有時效性，所以存儲時可以考慮冷熱存儲分離。

（3）數據查詢和分析。時序數據的查詢也具有顯著特點，一般會按照時間範圍讀取，最近的數據讀取頻率高，並且按照不同的時間粒度做聚合查詢，比如統計最近一周每天的數據量。

分析是依賴於查詢的，時序數據的分析通常是多維的，比如網頁點擊流量、從哪個網站、來自哪個IP、點擊頻率等維度眾多，取決於具體場景。而時序數據也非常適合數據挖掘，利用歷史預測未來。

（4）數據刪除。這裡的刪除並不是針對單條數據的，而是對特定時間範圍內的批量數據進行過期處理。因為時序數據具有時效性，歷史數據通常不再具有價值，不管是定時刪除還是手動刪除，都代表著其短暫的生命周期的結束。

2.2 時序數據管理系統目標

根據時序數據的特點和場景，我們需要一個能滿足以下目標的時序數據管理平臺：

• 高吞吐寫入：千萬、上億數據的秒級實時寫入 & 持續高並發寫入。
• 無更新操作：數據大多表徵設備狀態，寫入後無需更新。
• 海量數據存儲：從TB到PB級。
• 高效實時的查詢：按不同維度對指標進行統計分析，存在明顯的冷熱數據，一般只會頻繁查詢近期數據
• 高可用
• 可擴展性
• 易於使用
• 易於維護

2.3技術選型

說到資料庫，大家第一個想到的肯定是MySQL、Oracle等傳統的已經存在很多年的關係型資料庫。當然關係模型依然有效且實用。對於小數據量（幾百萬到幾千萬），MySQL是可以搞定的，再大一些就需要分庫分表解決了。對時序數據一般按照時間分表，但是這對外部額外設計和運維的工作提出了高要求。顯然，這不能滿足大數據場景，所以幾乎沒有人選擇這種方案。

縱觀db-engine上排名前十的時序資料庫[2]，排除商用的，剩下開源的選擇並不多。接下來介紹幾款比較流行的時序資料庫。

圖2 db-engine時序資料庫排名

（1）OpenTSDB。OpenTSDB開源快10年了，屬於早期的解決方案。因為其基於Hadoop和HBase開發的索引，所以具有海量數據的存儲能力，也號稱每秒百萬級別的寫入速度。但同樣因為其依賴的Hadoop生態太重, 運維成本很高，不夠簡潔與輕量；另一個缺點就是它基於HBase的key-value存儲方式，對於聚合查詢並不友好高效，HBase存在的問題也會體現出來。

圖3 OpenTSDB用戶界面

（2）InfluxDB。InfluxDB可以說是時序行業的典範了，其已經發展成為一個平臺，包括了時序數據應有的一切：從數據存儲到界面展示。然而，InfluxDB雖然開源了其核心代碼，但重要的集群功能只有企業版才提供[3], 而企業版並不是免費的。很多大公司要麼直接使用，要麼自己開發集群功能。

圖4 InfluxDB各版本支持的功能

（3）TDengine。TDengine是濤思團隊開發的一個高效存儲、查詢和分析時序大數據的平臺，其創始人陶建輝年近5旬，依然開發出了這個資料庫。

TDengine的定位是物聯網、車聯網、運維監測等時序數據，其設計也是專門針對每個設備。每個採集點一張表，比如空氣監測站有1000萬個，那麼就建1000萬個表，為了對多個採集點聚合查詢，又提出了超表的概念，將同類型的採集點表通過標籤區分，結構一樣。這種設計確實非常有針對性，雖然限制了範圍，但極大提高了效率，根據其官方的測試報告[4], 其聚合查詢速度是InfluxDB的上百倍，CPU、內存和硬碟消耗卻更少。

圖5 濤思團隊給出的不同時序資料庫性能對比

TDengine無疑是時序資料庫的一朵奇葩，加上在不久前開源了其集群功能[5]，受到了更多用戶青睞。當我們選型時其還沒有開源集群功能，後續也會納入觀察之中。

（4）ClickHouse。ClickHouse（之後簡稱CK）是一個開源的大數據分析資料庫，也是一個完整的DBMS。CK無疑是OLAP資料庫的一匹黑馬，開源不到4年，GitHub上的star數已經超過12k（InfluxDB也不過19k+），而它們的fork數卻相差不大。

CK是俄羅斯的搜尋引擎公司yandex開源的，最初是為了分析網頁點擊的流量，所以叫Click，迭代速度很快，每個月一版，開發者500+，很多都是開源共享者，社區非常活躍。

CK是一個通用的分析資料庫，並不是為時序數據設計的，但只要使用得當，依然能發揮出其強大的性能。

三、CK原理介紹

要利用CK的優勢，首先得知道它有哪些優勢，然後理解其核心原理。根據我們的測試結果，對於27個欄位的表，單個實例每秒寫入速度接近200MB，超過400萬條數據/s。因為數據是隨機生成的，對壓縮並不友好。

而對於查詢，在能夠利用索引的情況下，不同量級下（百萬、千萬、億級）都能在毫秒級返回。對於極限情況：對多個沒有索引的欄位做聚合查詢，也就是全表掃描時，也能達到400萬條/s的聚合速度。

3.1 CK為什麼快

可以歸結為選擇和細節，選擇決定方向，細節決定成敗。

CK選擇最優的算法，比如列式壓縮的LZ4[6]；選擇著眼硬體，充分利用CPU和分級緩存；針對不同場景不同處理，比如SIMD應用於文本和數據過濾；CK的持續迭代非常快，不僅可以迅速修復bug，也能很快納入新的優秀算法。

3.2 CK基礎

（1）CK是一個純列式存儲的資料庫，一個列就是硬碟上的一個或多個文件（多個分區有多個文件），關於列式存儲這裡就不展開了，總之列存對於分析來講好處更大，因為每個列單獨存儲，所以每一列數據可以壓縮，不僅節省了硬碟，還可以降低磁碟IO。

（2）CK是多核並行處理的，為了充分利用CPU資源，多線程和多核必不可少，同時向量化執行也會大幅提高速度。

（3）提供SQL查詢接口，CK的客戶端連接方式分為HTTP和TCP，TCP更加底層和高效，HTTP更容易使用和擴展，一般來說HTTP足矣，社區已經有很多各種語言的連接客戶端。

（4）CK不支持事務，大數據場景下對事務的要求沒這麼高。

（5）不建議按行更新和刪除，CK的刪除操作也會轉化為增加操作，粒度太低嚴重影響效率。

3.3 CK集群

生產環境中通常是使用集群部署，CK的集群與Hadoop等集群稍微有些不一樣。如圖6所示，CK集群共包含以下幾個關鍵概念。

圖6 CK集群示例

（1）CK實例。可以一臺主機上起多個CK實例，埠不同即可，也可以一臺主機一個CK實例。

（2）分片。數據的水平劃分，例如隨機劃分時，圖5中每個分片各有大約一半數據。

（3）副本。數據的冗餘備份，同時也可作為查詢節點。多個副本同時提供數據查詢服務，能夠加快數據的查詢效率，提高並發度。圖5中CK實例1和示例3存儲了相同數據。

（4）多主集群模式。CK的每個實例都可以叫做副本，每個實體都可以提供查詢，不區分主從，只是在寫入數據時會在每個分片裡臨時選一個主副本，來提供數據同步服務，具體見下文中的寫入過程。

3.4 CK分布式引擎

要實現分片的功能，需要分布式引擎。在集群情況下，CK裡的表分為本地表和分布式表，下面的兩條語句能夠創建一個分布式表。注意，分布式表是一個邏輯表，映射到多個本地表。

create table t_local on cluster shard2_replica2_cluster(t Datetime, id UInt64)

ENGINE=ReplicatedMergeTree('/clickhouse/tables/{shard}/t_local','{replica}')

PARTITION BY toYYYYMM(t)

ORDER BY id

create table t on cluster shard2_replica2_cluster (t Datetime, id UInt64)

ENGINE=Distributed(shard2_replica2_cluster,default,t_local,id)

這裡的t_local就是本地表，t就是分布式表。ReplicatedMergeTree是實現副本同步的引擎，參數可以先忽略。Distributed引擎就是分布式引擎，參數分別為：集群名，資料庫名，本地表名，分片鍵（可以指定為rand()隨機數）。

分布式引擎在寫入和查詢過程中都充當著重要的角色，具體過程見下面。

3.5 CK寫入過程

根據使用的表引擎不同，寫入過程是不同的，上文的建表方式是比較常規的做法，按照上面的建表語句，需要同時開啟內部複製項。

<shard2_replica2_cluster>

<shard>

<weight>1</weight>

<internal_replication>true</internal_replication>

<replica>

…

</replica>

<replica>

…

</replica>

</shard>

…

寫入2條數據：insert into t values(now(), 1), (now(),2)，如圖7所示，寫入過程分為2步：分布式寫入和副本同步。

圖7 CK寫入過程

（1）分布式寫入

1）客戶端會選擇集群裡一個副本建立連接，這裡是實例1。寫入的所有數據先在實例1完成寫入，根據分片規則，屬於01分片的寫入實例1本地，屬於02分片的先寫入一個臨時目錄，然後向實例2（shard02的主副本）建立連接，發送數據到實例2。

2）實例2接收到數據，寫入本地分區。

3）實例1返回寫入成功給客戶端（每個分片寫入一個副本即可返回，可以配置）。

（2）副本同步

同步的過程是需要用到ZK的，上面建表語句的ReplicatedMergeTree第一個參數就是ZK上的路徑。創建表的時候會有一個副本選舉過程，一般先起的會成為主副本，副本的節點信息會註冊到ZK，ZK的作用只是用來維護副本和任務元數據以及分布式通信，並不傳輸數據。副本一旦註冊成功，就開始監聽/log下的日誌了，當副本上線，執行插入時會經過以下過程：

1）實例1在寫入本地分區數據後，會發送操作日誌到ZK的/log下，帶上分區名稱和源主機（實例1的主機）。

2）01分區的其他副本，這裡就實例3，監聽到日誌的變化，拉取日誌，創建任務，放入ZK上的執行隊列/queue（這裡都是異步進行），然後再根據隊列執行任務。

3）執行任務的過程為：選擇一個副本（數據量最全且隊列任務最少的副本），建立到該副本（實例1）的連接，拉取數據。

注意，使用副本引擎卻不開啟內部複製是不明智的做法，因為數據會重複寫，雖然數據校驗可以保證數據不重複，但增加了無畏的開銷。

3.6 CK查詢過程

查詢的是分布式表，但要定位到實際的本地表，也就是副本的選擇，這裡有幾種選擇算法，默認採用隨機選擇。響應客戶端查詢請求的只會有一個副本，但是執行過程可能涉及多個副本。比如：select count(*) from t。因為數據是分布在2個分片的，只差一個副本不能得到全部結果。

圖8 CK多實例查詢過程

3.7 CK中重要的索引引擎

CK核心的引擎就是MergeTree，在此之上產生了很多附加引擎，下面介紹幾種比較常用的。

（1）ReplacingMergeTree。為了解決MergeTree主鍵可以重複的特點，可以使用ReplacingMergeTree，但也只是一定程度上不重複：僅僅在一個分區內不重複。使用方式參考：https://clickhouse.tech/docs/en/engines/table-engines/mergetree-family/replacingmergetree/

（2）SummingMergeTree。對於確定的group by + sum查詢，若比較耗時，那麼可以建SummingMergeTree, 按照order by的欄位進行聚合或自定義聚合欄位，其餘欄位求和。

（3）AggregatingMergeTree。聚合顯然是分析查詢的重點，一般使用聚合MergeTree都會結合物化視圖，在插入數據時自動同步到物化視圖裡，這樣直接查詢物化視圖中聚合的結果即可。

（4）ReplicatedXXXMergeTree。在所有引擎前加一個Replicated前綴，將引擎升級為支持副本功能。

（5）物化視圖。物化視圖就是將視圖SQL查詢的結果存在一張表裡，CK裡特殊的一點是：只有insert的數據才能進入觸發視圖查詢，進入視圖表，分布式情況下同步過去的數據是不會被觸發的，為了在分布式下使用物化視圖，可以將物化視圖所依賴的表指定為分布式表。

四、CK與時序的結合

在了解了CK的基本原理後，我們看看其在時序數據方面的處理能力。

（1）時間：時間是必不可少的，按照時間分區能夠大幅降低數據掃描範圍；

（2）過濾：對條件的過濾一般基於某些列，對於列式存儲來說優勢明顯；

（3）降採樣：對於時序來說非常重要的功能，可以通過聚合實現，CK自帶時間各個粒度的時間轉換函數以及強大的聚合能力，可以滿足要求；

（4）分析挖掘：可以開發擴展的函數來支持。

另外CK作為一個大數據系統，也滿足以下基礎要求：

（1）高吞吐寫入；

（2）海量數據存儲：冷熱備份，TTL；

（3）高效實時的查詢；

（4）高可用；

（5）可擴展性：可以實現自定義開發；

（6）易於使用：提供了JDBC和HTTP接口；

（7）易於維護：數據遷移方便，恢復容易，後續可能會將依賴的ZK去掉，內置分布式功能。

因此，CK可以很方便的實現一個高性能、高可用的時序數據管理和分析系統。下面是關鍵點的詳細介紹。

4.1 時序索引與分區

時序查詢場景會有很多聚合查詢，對於特定場景，如果使用的非常頻繁且數據量非常大，我們可以採用物化視圖進行預聚合，然後查詢物化視圖。但是，對於一個通用的分析平臺，查詢條件可以隨意改變的情況下，使用物化視圖的開銷就太大了，因此我們目前並沒有採用物化視圖的方式，而是採用原始的表。物化視圖的方案後續將會進一步驗證。

下面給出的是JUST建時序表的語法格式：第一個括號為TAG欄位，第二個括號為VALUE欄位（必須是數值型），大括號是對底層存儲的特殊配置，這裡主要是CK的索引和參數。除了用戶指定的欄位外，還有一個隱含的time欄位，專為時序保留。

create table my_ts_table as ts (

tag1 string,

tag2 String [:primarykey][:comment=』描述』]

)

(

value1 double,

value2 double

)

在JUST底層，對應了CK的2張表（一張本地表，一張分布式表），默認會根據time分區和排序，如下面的一個例子：

create table airquality as ts (

name string,

city String

)

(

PM10 double,

PM25 double

)

實際對應的CK建表語句為：

CREATE TABLE just.username_dbname_airquality_local

(

`id` Int32,

`oid`Int32,

`name`String,

`city`String,

`time`DateTime,

`PM10`Float64,

`PM25`Float64

)

ENGINE =ReplicatedMergeTree('/clickhouse/tables/{shard}/24518511-2939-489b-94a8-0567384d927d','{replica}')

ORDER BY (time)

SETTINGS index_granularity = 8192

PARTITION BY toYYYYMM(time)

CREATE TABLE just.wangpeng417_test_airquality

(

`id` Int32,

`oid`Int32,

`name`String,

`city`String,

`time`DateTime,

`PM10`Float64,

`PM25`Float64

)

ENGINE = Distributed('just_default', 'just', ' username_dbname_airquality_local',rand())

這樣保證在使用時間範圍查詢時可以利用到索引，假如還有其他按照TAG的查詢條件，還可以自定義索引和排序欄位[LL1] （CK規定索引欄位一定是排序欄位的前綴）。

在不同場景下，還是需要根據數據量和數據特點來選擇索引分區和索引粒度。根據實驗測試，假如在我們環境裡CK每秒可以掃描1GB數據量，再乘以1-10倍的壓縮比，那麼一個分區的數據量應該大於千萬到億級別可以保證較優的速度，CK本身是多線程查詢的，可以保證同時對每個分區查詢的隔離性。但是根據查詢的場景，比如最多查到一個月，但大部分情況都是查一周，那麼分區精確到周可能更好，這是個綜合權衡的過程。

4.2 部署與高可用

在JUST中，高可擴展性和高可用性是我們的追求。為實現高可擴展性，我們對數據進行水平分片；為了實現高可用性，我們對每個分片存儲至少兩個副本。

關於集群部署，最小化的情況是2臺機器，這會產生2種情況1）交叉副本；2）一主一備；如圖9所示：

圖9 兩種副本的情形

這2種方案對查詢和寫入影響的實驗結果如圖10所示:

圖10 兩種副本的寫入和查詢結果對比

實驗結果表明：寫入速度（橫坐標為寫入示例數，縱坐標為速度MB/s）在達到極限時是差不多的，而查詢性能（橫坐標為SQL編號,SQL語句見附錄1，縱坐標為耗時，單位為秒）對於簡單查詢差別不大，但是對於佔用大量資源的複雜查詢，一主一備更加高效。因為CK的強悍性能是建立在充分利用CPU的基礎上，在我們的測試中，裸機情況下CPU達到90%以上非常頻繁，如果有單獨的機器部署CK，那麼無可厚非能夠充分利用機器資源。但在我們的環境中，與其他大數據平臺共有機器，就需要避免CK佔用過多資源，從而影響其他服務，於是我們選擇docker部署。docker容器部署也有開源的基於k8s的實現：clickhouse-operator，對於小型環境，可以選擇手動配置，通過簡單的腳本即可實現自動化部署。

基於以上測試結論，為了保證服務高可用，CK集群和數據冗餘是必不可少的，我們的方案是保證至少2個副本的情況下，分片數儘量多，充分利用機器，且每個機器有且僅有一個CK實例。於是就有了以下分片數與副本數的公式：

其中f(n)代表當有n臺機器時，部署的分布情況，n>=2。f(2) = (1, 2)表示2臺機器採用1個分片2個副本部署的策略，f(3)=(1, 3)表示3臺機器時1個分片3個副本部署策略，f(4)=(2, 2)表示4臺機器使用2個分片，每個分片2個副本，以此類推。

4.3 動態擴容

隨著數據量增加，需要擴展節點時，可以在不停機的情況下動態擴容，主要利用的是分片之間的權重關係。

這裡擴容分為2種情況：

（1）增加副本：只需要修改配置文件，增加副本實例，數據會自動同步，因為CK的多主特性，副本也可以當作查詢節點，所以可以分擔查詢壓力；

（2）增加分片：增加分片要麻煩點，需要根據當前數據量、增加數據量計算出權重，然後在數據量達到均衡時將權重修改回去

假如開始時我們只有1個分片，已經有100條數據了

<test_extend>

<shard>

<weight>1</weight>

<internal_replication>true</internal_replication>

<replica>

<host>10.220.48.106</host>

<port>9000</port>

</replica>

<replica>

<host>10.220.48.105</host>

<port>9000</port>

</replica>

</shard>

</test_extend>

現在要再加入一個分片，那麼權重的計算過程如下（為了簡化忽略這個期間插入的數據）：

假如我們打算再插n條數據時，集群數據能夠均衡，那麼每個shard有(n+100)/2 條，現在shard01有100條，設權重為 w1、w2，那滿足公式：n * (w2/(w1+w2)) = (n+100)/2 ，其中n>100, 所以，假如 w1=1，n=200，那麼 w2=3。

所以，將配置修改如下：

<test_extend>

<shard>

<weight>1</weight>

<internal_replication>true</internal_replication>

<replica>

<host>10.220.48.106</host>

<port>9000</port>

</replica>

<replica>

<host>10.220.48.105</host>

<port>9000</port>

</replica>

</shard>

<shard>

<weight>3</weight>

<internal_replication>true</internal_replication>

<replica>

<host>10.220.48.103</host>

<port>9000</port>

</replica>

</shard>

</test_extend>

等到數據同步均勻後再改回1：1

4.4系統介紹與不足

JUST時序分析底層使用了CK作為存儲查詢引擎，並開發了可復用的可視化分析界面，歡迎訪問https://just.urban-computing.cn/進行體驗。

圖11 JUST時序分析模塊示意圖

用戶可以使用統一的查詢界面建立時序表，然後導入數據，切換到時序分析模塊進行可視化查詢。

圖12 JUST建立時序表示意圖

目前提供的查詢功能主要有：按時間查詢、按TAG過濾，在數據量很多的情況下，可以按照大一些的時間粒度進行降採樣，查看整個數據的趨勢，同時提供了線性、拉格朗日等缺失值填補功能。

分析挖掘部分主要是按照特定值和百分比過濾，以及一些簡單的函數轉換。

目前時序模塊的功能還比較簡陋，對於時序數據的SQL查詢支持還不夠完備。未來還有集成以下功能：

（1）接入實時數據；

（2）針對複雜查詢，面板功能可以採用聚合引擎預先聚合；

（3）更完善的分析和挖掘功能；

（4）對數據的容錯與校驗處理；

（5）與JUST一致的SQL查詢支持。

參考連結：

[1]https://en.wikipedia.org/wiki/Time_series

[2]https://db-engines.com/en/ranking/time+series+dbms

[3]https://www.influxdata.com/blog/influxdb-clustering/

[4]https://www.taosdata.com/downloads/TDengine_Testing_Report_cn.pdf

[5]https://www.taosdata.com/blog/2020/08/03/1703.html

[6]lz4.LZ4[EB/OL].https://lz4.github.io/lz4/,2014-08-10.

[7]https://clickhouse.tech/docs/en/engines/table-engines/mergetree-family/mergetree/

附錄：

-- SQL1：存在聚合函數

select

avg(rainfall)

from

t_air_one_one_dist_1;

-- SQL2：存在聚合函數以及排序

select

county_name,

count(*) ascnt

from

t_air_one_one_dist_1

group by

county_name

order by

cnt desc,

county_name

limit

10;

-- SQL3：存在聚合函數以及排序

select

county_name,

avg(rainfall)as cnt

from

t_air_one_one_dist_1

group by

county_name

order by

cnt desc,

county_name

limit

10;

-- SQL4：存在聚合函數並且含有having子句

select

county_name,

count(*)

from

t_air_one_one_dist_1

group by

county_name

having

count(*) >1

limit

10;

-- SQL5：存在聚合函數

select

sum(rainfall)

from

t_air_one_one_dist_1;

-- SQL6：存在聚合函數、排序

select

avg(rainfall)as cnt

from

t_air_one_one_dist_1

group by

city_name,

county_name

order by

cnt desc

limit

10;

-- SQL7：存在dist

select

wind_speed,

avg(rainfall)as cnt,

count(distinct(county_name)) as avg1

from

t_air_one_one_dist_1

group by

city_name,

county_name,

wind_speed

order by

cnt desc,

avg1

limit

10;