背景

如果你是用戶，當你使用抖音、小紅書的時候，假如平臺能根據你的屬性、偏好、行為推薦給你感興趣的內容，那就能夠為你節省大量獲取內容的時間。

如果你是商家，當你要進行廣告投放的時候，假如平臺推送的用戶都是你潛在的買家，那你就可以花更少的錢，帶來更大的收益。

這兩者背後都有一項共同的技術支撐，那就是用戶畫像。

業務能力

京東科技畫像系統，提供標準的畫像功能服務，包含標簽市場、人群管理、數據服務、標簽管理等，可以將用戶分群服務於其他各個業務系統。

挑戰

• 數據量大

目前平臺擁有百億+的用戶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來實現高效、高壓縮的存儲策略，但整個進化過程是一步步遞進完成的。

數據服務

人群加工完成後要對外提供命中服務，為了支撐高併發高性能的調用需求，人群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著手建設數據實時化，目前已經完全做到了人群命中實時計算。

作者：京東科技 梁發文

來源：京東雲開發者社區 轉載請註明來源