【實用+乾貨】如何使用Clickhouse搭建百億級人物誌平臺看這一篇就夠了
背景
如果你是使用者，當你使用抖音、小紅書的時候，假如平臺能根據你的屬性、偏好、行為推薦給你感興趣的內容，那就能夠為你節省大量獲取內容的時間。
如果你是商家，當你要進行廣告投放的時候，假如平臺推播的使用者都是你潛在的買家，那你就可以花更少的錢，帶來更大的收益。
這兩者背後都有一項共同的技術支撐，那就是人物誌。
業務能力
京東科技畫像系統，提供標準的畫像功能服務，包含標籤市場、人群管理、資料服務、標籤管理等，可以將使用者分群服務於其他各個業務系統。
挑戰
- 資料量大
目前平臺擁有百億+的使用者ID、5000+的標籤，單個人群包內的使用者數量可達數十億級，每天更新的人群也有2W多個。
- 計算複雜
標籤圈選的條件複雜，底層依賴的資料量級較高，人群計算需要進行大量的交併差計算。
- 查詢時間短
如果人群數預估、人群建立的耗時較長，對業務方的影響較大。
- 資料儲存成本高
大量的人群資料儲存需要高昂的儲存成本。
- 資料查詢量大、效能要求高
大促期間介面呼叫量高達百萬QPS，介面響應要求要在40毫秒以內，而且要支援批次人群呼叫。
壓縮的Bitmap
Bitmap 是一個二進位制集合，用0或1標識某個值是否存在，使用Bitmap的特點和標籤、人群結果的結構高度契合，正常1億的人群包使用Bitmap儲存只需要50MB左右。
在求兩個集合的交併差運算時，不需要遍歷兩個集合，只要對位進行與運算即可。無論是比較次數的降低（從 O(N^2) 到Ｏ(N) ），還是比較方式的改善（位運算），都給效能帶來巨大的提升。
從RoaringBitmap說起
RoaringBitmap(簡稱RBM)是一種高效壓縮點陣圖，本質上是將大塊的bitmap分成各個小桶，其中每個小桶在需要儲存資料的時候才會被建立，從而達到了壓縮儲存和高效能運算的效果。
在實際儲存時，先把64位元的數劃分成高32位元和低32位元，建立一個我們稱為Container的容器，同樣的再分別為高低32位元建立高16位元和低16位元的Container，最終可以通過多次二分查詢找到offset所在的小桶。
選擇Clickhouse
- Clickhouse的特點
完備的資料庫管理功能，包括DML（資料操作語言）、DDL（資料定義語言）、許可權控制、資料備份與恢復、分散式計算和管理。
列式儲存與資料壓縮: 資料按列儲存，在按列聚合的場景下，可有效減少查詢時所需掃描的資料量。同時，按列儲存資料對資料壓縮有天然的友好性（列資料的同類性），降低網路傳輸和磁碟 IO 的壓力。
關係模型與SQL: ClickHouse使用關係模型描述資料並提供了傳統資料庫的概念（資料庫、表、檢視和函數等）。與此同時，使用標準SQL作為查詢語言，使得它更容易理解和學習，並輕鬆與第三方系統整合。
資料分片與分散式查詢: 將資料橫向切分，分佈到叢集內各個伺服器節點進行儲存。與此同時，可將資料的查詢計算下推至各個節點並行執行，提高查詢速度。
- 為什麼選擇Clickhouse
分析效能高：在同類產品中，ClickHouse分析效能遙遙領先，複雜的人群預估SQL可以做到秒級響應。
簡化開發流程：關係型資料庫和SQL對於開發人員有天然的親和度，使得所有的功能開發完全SQL化，支援JDBC，降低了開發和維護的成本。
開源、社群活躍度高：版本迭代非常快，幾乎幾天到十幾天更新一個小版本，發展趨勢迅猛。
支援壓縮點陣圖：資料結構上支援壓縮點陣圖，有完善的Bitmap函數支撐各種業務場景。
- Clickhouse的部署架構
採用分散式多分片部署，每個分片保證至少有2個節點互為主備，來達到高效能、高可用的目的。
分片和節點之間通過Zookeeper來儲存後設資料，以及互相通訊。這樣可以看出Clickhouse本身是對Zookeeper是強依賴的，所以通常還需要部署一個3節點的高可用Zookeeper叢集。
- 分散式表和本地表
本地表指每個節點的實際儲存資料的表，有兩個特點:
1、每個節點維護自己的本地表；
2、每個本地表只管這個節點的資料。
本地表每個節點都要建立，CK通常是會按自己的策略把資料平均寫到每一個節點的本地表，本地資料本地計算，最後再把所有節點的結果彙總到一起。
通常我們也可以通過DDL里加上ON CLUSTER [叢集模式] 這樣的形式在任意節點執行即可在全部節點都建立相同的表。
例如通過ON CLUSTER模式執行DDL語句在每個節點建立名為[test]的庫，其中[default]為建立叢集時設定的叢集名稱
CREATE DATABASE test ON CLUSTER default
通常我們可以在應用裡通過JDBC在每個節點執行SQL得到結果後，再在應用內進行聚合，要注意的是像平均值這樣的計算，只能是通過先求SUM再求COUNT來實現了，直接使用平均值函數只能得到錯誤的結果。
分散式表是邏輯上的表、可以理解位檢視。比如要查一張表的全量資料，可以去查詢分散式表，執行時分發到每一個節點上，等所有節點都執行結束再在一個節點上彙總到一起(會對單節點造成壓力)。
查詢分散式表時，節點之間通訊也是依賴zk，會對zk造成一定的壓力。
同樣的分散式表如果需要每個節點都能查詢，就得在每一個節點都建表，當然也可以使用ON CLUSTER模式來建立。
例如為test.test_1在所有節點建立分散式表:
CREATE TABLE test.test_1_dist ON CLUSTER default
as test.test_1 ENGINE = Distributed('default','test','test_1',rand());
- 對Bitmap的支援
a、建立包含有Bitmap列的表
CREATE TABLE cdp.group ON CLUSTER default
(
`code` String,
`version` String,
`offset_bitmap` AggregateFunction(groupBitmap, UInt64)
)
ENGINE = ReplicatedMergeTree('/clickhouse/tables/cdp/{shard}/group_1',
'{replica}')
PARTITION BY (code)
ORDER BY (code)
SETTINGS storage_policy = 'default',
use_minimalistic_part_header_in_zookeeper = 1,
index_granularity = 8192;
b、Bitmap如何寫入CK
通常有2種方式來寫入Bitmap：
1、第一種通過INSERT .... SELECT...來執行INSERT SQL語句把明細資料中的offset轉為Bitmap
INSERT INTO cdp.group SELECT 'group1' AS code, 'version1' AS version, groupBitmapState(offset) AS offset_bitmap FROM tag.tag1 WHERE ....
2、在應用內生成Bitmap通過JDBC直接寫入
<dependency>
<!-- please stop using ru.yandex.clickhouse as it's been deprecated -->
<groupId>com.clickhouse</groupId>
<artifactId>clickhouse-jdbc</artifactId>
<version>0.3.2-patch11</version>
<!-- use uber jar with all dependencies included, change classifier to http for smaller jar -->
<classifier>all</classifier>
<exclusions>
<exclusion>
<groupId>*</groupId>
<artifactId>*</artifactId>
</exclusion>
</exclusions>
</dependency>
String sql = "INSERT INTO cdp.group SELECT code,version,offset_bitmap FROM input('code String,version String,offset_bitmap AggregateFunction(groupBitmap,UInt64)')";
try (PreparedStatement ps = dataSource.getConnection().prepareStatement(
sql)) {
ps.setString(1, code); // col1
ps.setString(2, uuid); // col2
ps.setObject(3, ClickHouseBitmap.wrap(bitmap.getSourceBitmap(), ClickHouseDataType.UInt64)); // col3
ps.addBatch();
ps.executeBatch();
}
c、從CK讀取Bitmap
直接讀取：
ClickHouseStatement statement =null;
try{
statement = connection.createStatement();
ClickHouseRowBinaryInputStream in = statement.executeQueryClickhouseRowBinaryStream(sql);
ClickHouseBitmap bit = in.readBitmap(ClickHouseDataType.UInt64);
Roaring64NavigableMap obj =(Roaring64NavigableMap) bit.unwrap();
returnnewExtRoaringBitmap(obj);
}catch(Exception e){
log.error("查詢點陣圖錯誤 [ip:{}][sql:{}]", ip, sql, e);
throw e;
}finally{
DbUtils.close(statement);
}
上面的方法直接讀取Bitmap會大量佔用應用記憶體，怎麼進行優化呢？ 我們可以通過Clickhouse把Bitmap轉成列，通過流式讀取bitmap裡的offset，在應用裡建立Bitmap
privatestatic String SQL_WRAP="SELECT bitmap_arr AS offset FROM (SELECT bitmapToArray(offset_bitmap) AS bitmap_arr FROM (SELECT ({}) AS offset_bitmap LIMIT 1)) ARRAY JOIN bitmap_arr";
publicstatic Roaring64NavigableMap getBitmap(Connection conn, String sql){
Statement statement =null;
ResultSet rs =null;
try{
statement = conn.createStatement(
ResultSet.TYPE_FORWARD_ONLY,
ResultSet.CONCUR_READ_ONLY);
String exeSql = ATool.format(SQL_WRAP, sql);
rs = statement.executeQuery(exeSql);
Roaring64NavigableMap bitmap =newRoaring64NavigableMap();
while(rs.next()){
Long offset = rs.getLong(1);
if(offset !=null){
bitmap.add(offset);
}
}
return bitmap;
}catch(Exception e){
log.error("從CK讀取bitmap錯誤, SQL:{}", sql, e);
}finally{
ATool.close(rs, statement);
}
returnnull;
}
京東科技CDP
- 整體架構
- 資料流轉
- 進一步優化
1. 平行計算
前面提到了Clickhouse通常把資料按照分片數把資料拆分成n份，只要我們保證相同使用者id的資料在每一張本地表中都在同一個節點，那麼我們多表之間進行JOIN計算時只需要每一個節點的本地表之間進行計算，從而達到了平行計算的效果。
為了達到這個目的，那必須要從開始的明細表就要通過一定的策略進行切分，客製化什麼樣的切分策略呢？這就要從RoaringBitmap的特性和機制來考慮。
2. 提高Bitmap在各節點的壓縮率
標籤和人群的最終結果資料都是用RoaringBitmap來儲存的，如果每個Bitmap儲存的小桶數量越少，那麼計算和儲存的成本就會更小，使用哪種策略來切分就變得至關重要。
先按照RoaringBitmap的策略將資料按照2的16次方為單位，切分成多個小桶，然後為小桶進行編號，再按編號取餘來切分。這樣同一個RoaringBitmap小桶中的offset只會在1個分片上，從而達到了減少小桶個數的目的。
//把資料拆分成10個分片，計算每個offset再哪個分片上
long shardIndex = (offset >>> 16) % 10
再進一步向上看，明細表如果也保持這樣的分片邏輯，那麼從明細轉成Bitmap後，Bitmap自然就是高壓縮的。
3.標籤Bitmap表
標籤明細資料怎麼存，嚴重關係到人群計算的效率，經過長期的探索和優化，最終通過按列舉分組來加工出標籤Bitmap來實現高效、高壓縮的儲存策略，但整個進化過程是一步步遞進完成的。
| 階段 | 方案 | 優勢 | 劣勢和瓶頸 |
|---|---|---|---|
| 第一階段 | 標籤寬表(單表) | 查詢速度快、支援複雜條件 | 1、單表壓力大 2、合併單表、單表寫的時間過長，出錯就要全量重推 3、儲存成本高 |
| 第二階段 | 標籤窄表(多表) | 1、輕量化表管理靈活 2、標籤時效性提升較為突出 | 1、計算需要多表JOIN,計算時間長，還時常會有超時和計算不出結果的情況； |
| 第三階段 | 每個標籤按照列舉分組計算出標籤列舉的Bitmap儲存為中間結果表 | 解決了階段二計算時間長的問題，極大的節約了儲存和計算資源 | 表數量和資料分割區過多會增加ZK叢集的壓力； |
資料服務
人群加工完成後要對外提供命中服務，為了支撐高並行高效能的呼叫需求，人群Bitmap的儲存必須使用快取來儲存。
起初是把一個完整的Bitmap切分成了8份，儲存到8臺物理機的記憶體裡，每臺機器儲存了1/8的資料。這種儲存方式本身是沒有問題的，但面臨著運維複雜，擴容困難的問題。
那麼我們能不能使用Redis來儲存人群資料呢？經過探索發現Redis本身是不支援壓縮點陣圖的，當我們寫入一個2的64次方大小的offset時，就會建立一個龐大的Bitmap，佔用大量的記憶體空間。這時我們就想到了使用壓縮點陣圖的原理把點陣圖按照2的16次方大小切分成多個小桶，把大的Bitmap轉成小的Bitmap，在儲存時減去一定的偏移量，在讀取時在加上偏移量，那麼每一個小桶就是一個65536(2^16)大小的Bitmap。從而我們開發了一套完整的Agent程式來記錄後設資料資訊，進行路由和讀寫Redis，最終實現了Redis儲存壓縮Bitmap的目的。
long bucketIndex = offset / 65536L;
long bucketOffset = offset % 65536L;
//其中偏移量就是bucketIndex * 65536L
儲存時只需要把key+[bucketIndex]作為key,使用bucketOffset來setBit()。
進一步查檔案發現，Redis本身就是支援把Bitmap轉成位元組陣列後一次性寫入的，這樣又進一步的提升了資料寫入的效率。
總結
京東科技CDP畫像平臺通過對使用者分群，針對不同的使用者投放以不同形式的不同內容，實現千人千面的精準投放，最終實現使用者的增長。對外提供多樣的資料服務，服務於不同的業務，以支援精準行銷、精細化運營，智慧外呼等行銷場景。
隨著時代的發展，離線人群已經不能滿足日益增長的運營需求。從去年開始,CDP著手建設資料實時化，目前已經完全做到了人群命中實時計算。
作者：京東科技 樑發文
來源：京東雲開發者社群 轉載請註明來源