ck官方網站（JUST技術）

本次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 idcreate 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 = 8192PARTITION 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：14.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(*) ascntfrom t_air_one_one_dist_1group by county_nameorder by cnt desc, county_namelimit 10; — SQL3：存在聚合函數以及排序select county_name, avg(rainfall)as cntfrom t_air_one_one_dist_1group by county_nameorder by cnt desc, county_namelimit 10; — SQL4：存在聚合函數並且含有having子句select county_name, count(*)from t_air_one_one_dist_1group by county_namehaving count(*) >1limit 10; — SQL5：存在聚合函數select sum(rainfall)from t_air_one_one_dist_1; — SQL6：存在聚合函數、排序select avg(rainfall)as cntfrom t_air_one_one_dist_1group by city_name, county_nameorder by cnt desclimit 10; — SQL7：存在distselect wind_speed, avg(rainfall)as cnt, count(distinct(county_name)) as avg1from t_air_one_one_dist_1group by city_name, county_name, wind_speedorder by cnt desc, avg1limit 10;