在 ClickHouse 中连续计数项目数

Question

我正在 ClickHouse 中开发一个简单的 API，它使用特定密钥持续计算不同用户的数量。

此环境有 2 个表和 1 个物化视图：

• 第一个表init_table 重复接收批量数据。
• 第二个表 final_table 使用 user_id 并基于由 hash_id 和 item1 两个元素组成的键来计算不同用户的数量。
• 这个计算是从init_table 到final_table 触发的，带有物化视图。

这里是创建表和物化视图的代码：

-- Init table
-- Table where data is continuously inserted in batches
DROP TABLE IF EXISTS test_db.init_table;
CREATE TABLE test_db.init_table (
    `timestamp` DateTime DEFAULT now(),
    `hash_id` FixedString(32),
    `item1` UInt32,
    `user_id` UInt32,
    `data1` UInt32,
    `data2` String
)   ENGINE = MergeTree()
    PARTITION BY tuple()
    ORDER BY ( hash_id, item1 )
    SETTINGS index_granularity = 8192;

-- Final table
DROP TABLE IF EXISTS test_db.final_table;
CREATE TABLE test_db.final_table (
    `timestamp` DateTime,
    `hash_id` FixedString(32),
    `item1` UInt32,
    `nb_user` UInt32
)   ENGINE = ReplacingMergeTree( timestamp )
    PARTITION BY tuple()
    ORDER BY ( hash_id, item1 )
    SETTINGS index_granularity = 8192;

-- Automating calculation from init table to final table
DROP TABLE IF EXISTS test_db.final_table_mv;
CREATE MATERIALIZED VIEW test_db.final_table_mv TO test_db.final_table AS
    SELECT
        timestamp,
        hash_id,
        item1,
        uniqExact( hash_id ) as nb_user
    FROM test_db.init_table
    GROUP BY ( timestamp, hash_id, item1 );

本例中用于聚合数据的Engine为ReplacingMergeTree，参数为插入数据的时间戳。

数据插入查询：

-- Data insertion
INSERT INTO test_db.init_table (hash_id,item1,user_id,data1,data2) VALUES ('564D6CE91699BC0174BED61EBA966A55',1,4444,'gnr','fbj'), ('564D6CE91699BC0174BED61EBA966A55',1,1111,'fhi','jdi'), ('564D6CE91699BC0174BED61EBA966A55',1,3333,'hvn','fhi');
SELECT sleep(2);
INSERT INTO test_db.init_table (hash_id,item1,user_id,data1,data2) VALUES ('564D6CE91699BC0174BED61EBA966A55',1,4444,'gnr','fbj'), ('61215DE218CC92BD74D82D2511EAC4CC',1,4444,'jbj','dhi'), ('5CC905405307AA837D943C266C84ECE9',1,4444,'vhi','bjh');
SELECT sleep(2);
INSERT INTO test_db.init_table (hash_id,item1,user_id,data1,data2) VALUES ('5CC905405307AA837D943C266C84ECE9',1,1111,'bjd','dic'), ('564D6CE91699BC0174BED61EBA966A55',1,1111,'fhi','jdi'), ('19DC7D744DD74D4BD15C298C118E72B7',1,3333,'hfj','bjd'), ('564D6CE91699BC0174BED61EBA966A55',1,3333,'hvn','fhi'), ('BAB3B080B7DF54D0831DC077F203673A',1,3333,'jij','vbj'), ('DED51D04E97D621780FC54580A9DA77B',1,1111,'vbj','hcn');
SELECT sleep(2);
INSERT INTO test_db.init_table (hash_id,item1,user_id,data1,data2) VALUES ('564D6CE91699BC0174BED61EBA966A55',1,5555,'fbj','jdh'), ('8C48E3B8888EB3C37B269B2D6A2A5206',1,5555,'dhi','vjs'), ('DED51D04E97D621780FC54580A9DA77B',1,5555,'bjh','jks');
SELECT sleep(2);
INSERT INTO test_db.init_table (hash_id,item1,user_id,data1,data2) VALUES ('564D6CE91699BC0174BED61EBA966A55',1,6666,'dic','msk'), ('3E33205D3367E2B9A3DB2F73A8CEF077',1,6666,'jdi','xok'), ('702893A3E0A402776BFCC3E7A4BF5F77',1,6666,'hcn','lxs');

在init_table中插入一些数据集后，final_table中显示的用户数是user_id的聚合基于数据集而不是基于@的内容987654334@.

-- Testing data
-- Number of distinct user_id in the init_table
select count(distinct user_id) from test_db.init_table where hash_id = '564D6CE91699BC0174BED61EBA966A55'; 
-- n = 5 --> this should be the right answer

-- Content of the final_table filtering on hash_id 564D6CE91699BC0174BED61EBA966A55
select * from test_db.final_table where hash_id = '564D6CE91699BC0174BED61EBA966A55' order by timestamp;
-- timestamp                hash_id                                 item1   nb_user
-- 2020-07-24 07:19:26      '564D6CE91699BC0174BED61EBA966A55'      1       3
-- 2020-07-24 07:19:28      '564D6CE91699BC0174BED61EBA966A55'      1       1
-- 2020-07-24 07:19:31      '564D6CE91699BC0174BED61EBA966A55'      1       2
-- 2020-07-24 07:19:33      '564D6CE91699BC0174BED61EBA966A55'      1       1
-- 2020-07-24 07:19:36      '564D6CE91699BC0174BED61EBA966A55'      1       1

-- Result after merging the data
select * from test_db.final_table final where hash_id = '564D6CE91699BC0174BED61EBA966A55' order by timestamp;
-- timestamp                hash_id                                 item1   nb_user
-- 2020-07-24 07:19:36      '564D6CE91699BC0174BED61EBA966A55'      1       1

所以，我在这里得到的最终结果不是init_table 中存在的不同user_id 的数量，而是插入init_table 的最后一个数据集中不同user_id 的数量。

我想要在final_table 中是init_table 组中不同的user_id 的总数hash_id 和item1（密钥），像这样：

hash_id                                 item1   nb_user
'564D6CE91699BC0174BED61EBA966A55'      1       5

关于我们这里的数据集，init_table 中不同的user_id 总数为 5。

我还尝试使用其他一些引擎，例如MergeTree 和AggregatingMergeTree，但没有成功。我究竟做错了什么。你有什么建议吗？

Answer 1

看起来 threre 是 test_db.final_table_mv-table 中的错误：而不是 uniqExact( hash_id ) as nb_user 需要使用 uniqExact( user_id ) as nb_user.

一般来说，您的代码容易出错，因为在 ReplacingMergeTree 合并行之后可能会丢失一些聚合值。

---

我会通过使用一张原始数据表和一个计算 intermediate states 的聚合 MV 来简化您的代码。中间状态允许计算任何维度组合的聚合（参见下面的查询）。

CREATE TABLE init_table (
  /* borrow origin code */
) /* .. */;


CREATE MATERIALIZED VIEW aggregates_mv
ENGINE = AggregatingMergeTree()
PARTITION BY tuple()
ORDER BY (hash_id, item1) 
AS
SELECT  
  hash_id, 
  item1,
  uniqExactState(user_id) AS nb_user
FROM init_table
GROUP BY hash_id, item1;

查询：

SELECT hash_id, item1, uniqExactMerge(nb_user) AS nb_user
FROM aggregates_mv
GROUP BY hash_id, item1;
/*
┌─hash_id──────────────────────────┬─item1─┬─nb_user─┐
   ...
│ 564D6CE91699BC0174BED61EBA966A55 │     1 │       5 │
   ...
└──────────────────────────────────┴───────┴─────────┘
*/

SELECT hash_id, uniqExactMerge(nb_user) AS nb_user
FROM aggregates_mv
GROUP BY hash_id;
/*
┌─hash_id──────────────────────────┬─nb_user─┐
   ...
│ 564D6CE91699BC0174BED61EBA966A55 │       5 │
   ... 
└──────────────────────────────────┴─────────┘
*/

SELECT item1, uniqExactMerge(nb_user) AS nb_user
FROM aggregates_mv
GROUP BY item1;
/*
┌─item1─┬─nb_user─┐
│     1 │       5 │
└───────┴─────────┘
*/