Clickhouse聚合计算
原创大约 11 分钟
同步行为数据
除了实现对风控指标的预计算，Clickhouse还需要存储并分析用户行为数据。
首先在Clickhouse中创建如下三张表。
-- 创建数据库
:) CREATE DATABASE IF NOT EXISTS dwd;
-- 创建数据同步表
:) DROP TABLE IF EXISTS dwd.event_middle;
:) CREATE TABLE IF NOT EXISTS dwd.event_middle (
eid UInt64,
etype String,
ename String,
esource String,
etime DateTime,
userid UInt64,
aid UInt64,
aname String,
tid String,
tname String,
context String
) ENGINE = Kafka() SETTINGS
kafka_broker_list = '172.16.185.176:9092',
kafka_topic_list = 'eventbus',
kafka_group_name = 'clickhouse',
kafka_format = 'JSONEachRow';
-- 创建物化视图
:) DROP VIEW IF EXISTS dwd.event_materialized_view;
:) CREATE MATERIALIZED VIEW IF NOT EXISTS dwd.event_materialized_view
TO dwd.event_table
SELECT
userid, ename, etime
FROM dwd.event_middle;
-- 创建用户行为数据表
:) DROP TABLE IF EXISTS dwd.event_table;
:) CREATE TABLE IF NOT EXISTS dwd.event_table (
userid UInt64,
ename String,
etime DateTime
) ENGINE = MergeTree()
ORDER BY userId;dwd.event_middle：由于使用的是Kafka()引擎，因此所有从Kafka中同步过来的数据都被会保存到这张表中。dwd.event_materialized_view：从Kafka中同步过来的数据是不能直接使用的，需要通过物化视表或者物化视图做一次中转。dwd.event_table：这是真正存放同步过来的用户行为数据的表。
当通过Flume将/var/logs/eventbus.log日志文件上传到Kafka时，Clickhouse就会自动消费指定Topic的数据，并通过物化视图dwd.event_materialized_view将数据保存到dwd.event_table表中。
聚合行为路径
经过同步后，dwd.event_table表现有如下数据。
:) SELECT * FROM dwd.event_table;
┌─userid─┬───ename────┬────────etime─────────┐
| 3 | 登录 | 2024-01-01 04:12:00 |
| 2 | 登录 | 2024-01-01 04:20:00 |
| 2 | 收藏商品 | 2024-01-01 04:25:00 |
| 2 | 查看详情页 | 2024-01-01 04:30:00 |
| 2 | 领券 | 2024-01-01 04:45:00 |
| 3 | 领券 | 2024-01-01 04:42:00 |
| 1 | 登录 | 2024-01-01 05:12:00 |
| 1 | 登录 | 2024-01-01 05:13:00 |
| 3 | 收藏商品 | 2024-01-01 05:15:00 |
| 1 | 领券 | 2024-01-01 05:15:08 |
| 2 | 查看详情页 | 2024-01-01 05:16:08 |
| 1 | 查看详情页 | 2024-01-01 05:17:08 |
| 1 | 查看详情页 | 2024-01-01 05:20:00 |
| 1 | 浏览商品 | 2024-01-01 05:25:00 |
| 3 | 加入购物车 | 2024-01-01 05:25:00 |
| 2 | 加入购物车 | 2024-01-01 05:25:30 |
| 1 | 浏览商品 | 2024-01-01 05:30:30 |
| 1 | 浏览商品 | 2024-01-01 05:30:31 |
| 3 | 用券 | 2024-01-01 05:40:00 |
| 1 | 用券 | 2024-01-01 05:40:30 |
| 2 | 查看详情页 | 2024-01-01 05:45:00 |
| 2 | 查看详情页 | 2024-01-01 06:15:00 |
| 2 | 查看详情页 | 2024-01-01 06:25:00 |
| 2 | 查看详情页 | 2024-01-01 06:27:38 |
| 2 | 查看详情页 | 2024-01-01 06:35:00 |
| 2 | 用券 | 2024-01-01 06:40:00 |
└────────┴────────────┴──────────────────────┘它记录的是电商用户的行为序列，例如，登录、领券、浏览收藏、查看详情和下单等事件。
1. 转换数据表
现在，需要让Clickhouse从这些行为序列中找到那些可能的薅羊毛或黑灰产用户，然后封禁这些账号。
通常来说，在拿到数据时，都会先将它转变成数组、元组或Map键值对。
这里也先将它转变为数组。同时，为了便于后续分析，需要将userid单独提取出来。
:) SELECT groupArray((userid, (ename, etime, toUInt32(etime))))
FROM dwd.event_table;
┌─groupArray((userid, (ename, etime, toUInt32(etime))))─────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────┐
│ [(3,('登录','2024-01-01 04:12:00',1704053520)),(2,('登录','2024-01-01 04:20:00',1704054000)),(2,('收藏商品','2024-01-01 04:25:00',1704054300)),(2,('查看详情页','2024-01-01 04:30:00',1704054600)),(3,('领券','2024-01-01 04:42:00',1704055320)),(2,('领券','2024-01-01 04:45:00',1704055500)),(1,('登录','2024-01-01 05:12:00',1704057120)),(1,('登录','2024-01-01 05:13:00',1704057180)),(3,('收藏商品','2024-01-01 05:15:00',1704057300)),(1,('领券','2024-01-01 05:15:08',1704057308)),(2,('查看详情页','2024-01-01 05:16:08',1704057368)),(1,('查看详情页','2024-01-01 05:17:08',1704057428)),(1,('查看详情页','2024-01-01 05:20:00',1704057600)),(1,('浏览商品','2024-01-01 05:25:00',1704057900)),(3,('加入购物车','2024-01-01 05:25:00',1704057900)),(2,('加入购物车','2024-01-01 05:25:30',1704057930)),(1,('浏览商品','2024-01-01 05:30:30',1704058230)),(1,('浏览商品','2024-01-01 05:30:31',1704058231)),(3,('用券','2024-01-01 05:40:00',1704058800)),(1,('用券','2024-01-01 05:40:30',1704058830)),(2,('查看详情页','2024-01-01 05:45:00',1704059100)),(2,('查看详情页','2024-01-01 06:15:00',1704060900)),(2,('查看详情页','2024-01-01 06:25:00',1704061500)),(2,('查看详情页','2024-01-01 06:27:38',1704061658)),(2,('查看详情页','2024-01-01 06:35:00',1704062100)),(2,('用券','2024-01-01 06:40:00',1704062400))]|
└───────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────┘通过groupArray()函数，就把数据表转换成了类似于下面这样的单行数组数据，而数组中的元素则是元组。
[
(3,('登录','2024-01-01 04:12:00',1704053520)),
(2,('登录','2024-01-01 04:20:00',1704054000)),
(2,('收藏商品','2024-01-01 04:25:00',1704054300)),
......
]2. 数据过滤
Clickhouse虽然很强，但它一次能处理的数据量毕竟还是有限的。
:) SELECT
arrayFilter(
x -> x.2.2 >= '2024-01-01 00:00:00' AND x.2.2 <= '2024-01-01 23:59:59',
groupArray((userid, (ename, etime, toUInt32(etime))))
)
FROM dwd.event_table;因为示例数据量比较少，所以这里过滤其实意义不大。
但在实际生产环境中必须要这样处理，因为一不小心就会把Clickhouse给撑爆。
3. 数据排序
为了让转换后的数据能够保持一个自然的时间序列，因此最好给它排个序。
:) SELECT
arraySort(
x -> x.2.2,
arrayFilter(
x -> x.2.2 >= '2024-01-01 00:00:00' AND x.2.2 <= '2024-01-01 23:59:59',
groupArray((userid, (ename, etime, toUInt32(etime))))
)
)
FROM dwd.event_table;arraySort()根据Lambda表达式指定的数值进行升序排序。
在实际生产环境中，数据是从各个不同的上游系统发过来的。有的没有排序，有的有排序，即使有排过序的，但如果系统宕机，重启之后再补发前面的数据，那么保存到Clickhouse时数据的顺序也会错乱。
所以，为保险起见，最好也先给数据排个序再进行下一步的处理。
4. 用户分组
虽然数据已经按照要求保存在数组中了，但用户行为都是混在一起的，需要按userid进行分组。
:) WITH
arrayFilter(
x->x.2.2 >= '2024-01-01 00:00:00' AND x.2.2 <= '2024-01-01 23:59:59',
groupArray((userid, (ename, etime, toInt32(etime))))
) AS sorted_events
SELECT userid, sorted_events, '2024-01-01'
FROM dwd.event_table
GROUP BY userid
ORDER BY userid;
┌─userid─┬─sorted_events─────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────┐
│ 1 │ [(1,('登录','2024-01-01 05:12:00',1704057120)),(1,('登录','2024-01-01 05:13:00',1704057180)),(1,('领券','2024-01-01 05:15:08',1704057308)),(1,('查看详情页','2024-01-01 05:17:08',1704057428)),(1,('查看详情页','2024-01-01 05:20:00',1704057600)),(1,('浏览商品','2024-01-01 05:25:00',1704057900)),(1,('浏览商品','2024-01-01 05:30:30',1704058230)),(1,('浏览商品','2024-01-01 05:30:31',1704058231)),(1,('用券','2024-01-01 05:40:30',1704058830))] │ 2024-01-01 │
│ 2 │ [(2,('登录','2024-01-01 04:20:00',1704054000)),(2,('收藏商品','2024-01-01 04:25:00',1704054300)),(2,('查看详情页','2024-01-01 04:30:00',1704054600)),(2,('领券','2024-01-01 04:45:00',1704055500)),(2,('查看详情页','2024-01-01 05:16:08',1704057368)),(2,('加入购物车','2024-01-01 05:25:30',1704057930)),(2,('查看详情页','2024-01-01 05:45:00',1704059100)),(2,('查看详情页','2024-01-01 06:15:00',1704060900)),(2,('查看详情页','2024-01-01 06:25:00',1704061500)),(2,('查看详情页','2024-01-01 06:27:38',1704061658)),(2,('查看详情页','2024-01-01 06:35:00',1704062100)),(2,('用券','2024-01-01 06:40:00',1704062400))] │ 2024-01-01 │
│ 3 │ [(3,('登录','2024-01-01 04:12:00',1704053520)),(3,('领券','2024-01-01 04:42:00',1704055320)),(3,('收藏商品','2024-01-01 05:15:00',1704057300)),(3,('加入购物车','2024-01-01 05:25:00',1704057900)),(3,('用券','2024-01-01 05:40:00',1704058800))] │ 2024-01-01 │
└────────┴───────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────┘好了，现在用户已经按userid分在了各自的组，而且所有的行为序列，都是只跟自己有关的。
5. 保存数据
因为在后续分析中需要用到这些数据，所以要将它们临时保存到事先准备好的容器中。
-- 创建数据库
:) CREATE DATABASE IF NOT EXISTS dws;
-- 创建临时数据表
:) CREATE TABLE IF NOT EXISTS dws.event_sequence_temporary
(
userid UInt64,
sequence Array(Tuple(UInt64, Tuple(String, DateTime, UInt32))),
windowtime DateTime
) ENGINE = MergeTree()
ORDER BY userid;
-- 保存数据
:) INSERT INTO dws.event_sequence_temporary
(userid, sequence, windowtime)
SELECT *
FROM
(
WITH
arrayFilter(
x->x.2.2 >= '2024-01-01 00:00:00' AND x.2.2 <= '2024-01-01 23:59:59',
groupArray((userid, (ename, etime, toInt32(etime))))
) AS sorted_events
SELECT userid, sorted_events, '2024-01-01'
FROM dwd.event_table
GROUP BY userid
ORDER BY userid
);
:) SELECT * FROM dws.event_sequence_temporary;
┌─userid─┬─sequence───────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────┬──────────windowtime─┐
│ 1 │ [(1,('登录','2024-01-01 05:12:00',1704057120)),(1,('登录','2024-01-01 05:13:00',1704057180)),(1,('领券','2024-01-01 05:15:08',1704057308)),(1,('查看详情页','2024-01-01 05:17:08',1704057428)),(1,('查看详情页','2024-01-01 05:20:00',1704057600)),(1,('浏览商品','2024-01-01 05:25:00',1704057900)),(1,('浏览商品','2024-01-01 05:30:30',1704058230)),(1,('浏览商品','2024-01-01 05:30:31',1704058231)),(1,('用券','2024-01-01 05:40:30',1704058830))] │ 2024-01-01 00:00:00 │
│ 2 │ [(2,('登录','2024-01-01 04:20:00',1704054000)),(2,('收藏商品','2024-01-01 04:25:00',1704054300)),(2,('查看详情页','2024-01-01 04:30:00',1704054600)),(2,('领券','2024-01-01 04:45:00',1704055500)),(2,('查看详情页','2024-01-01 05:16:08',1704057368)),(2,('加入购物车','2024-01-01 05:25:30',1704057930)),(2,('查看详情页','2024-01-01 05:45:00',1704059100)),(2,('查看详情页','2024-01-01 06:15:00',1704060900)),(2,('查看详情页','2024-01-01 06:25:00',1704061500)),(2,('查看详情页','2024-01-01 06:27:38',1704061658)),(2,('查看详情页','2024-01-01 06:35:00',1704062100)),(2,('用券','2024-01-01 06:40:00',1704062400))] │ 2024-01-01 00:00:00 │
│ 3 │ [(3,('登录','2024-01-01 04:12:00',1704053520)),(3,('领券','2024-01-01 04:42:00',1704055320)),(3,('收藏商品','2024-01-01 05:15:00',1704057300)),(3,('加入购物车','2024-01-01 05:25:00',1704057900)),(3,('用券','2024-01-01 05:40:00',1704058800))] │ 2024-01-01 00:00:00 │
└────────┴────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────┴─────────────────────┘6. 展开数据
分组之后，再展开每个用户的行为序列。
:) SELECT events
FROM dws.event_sequence_temporary
ARRAY JOIN sequence AS events;
┌─events──────────────────────────────────────────────┐
│ (1,('登录','2024-01-01 05:12:00',1704057120)) │
│ (1,('登录','2024-01-01 05:13:00',1704057180)) │
│ (1,('领券','2024-01-01 05:15:08',1704057308)) │
│ (1,('查看详情页','2024-01-01 05:17:08',1704057428)) │
│ (1,('查看详情页','2024-01-01 05:20:00',1704057600)) │
│ (1,('浏览商品','2024-01-01 05:25:00',1704057900)) │
│ (1,('浏览商品','2024-01-01 05:30:30',1704058230)) │
│ (1,('浏览商品','2024-01-01 05:30:31',1704058231)) │
│ (1,('用券','2024-01-01 05:40:30',1704058830)) │
│ (2,('登录','2024-01-01 04:20:00',1704054000)) │
│ (2,('收藏商品','2024-01-01 04:25:00',1704054300)) │
│ (2,('查看详情页','2024-01-01 04:30:00',1704054600)) │
│ (2,('领券','2024-01-01 04:45:00',1704055500)) │
│ (2,('查看详情页','2024-01-01 05:16:08',1704057368)) │
│ (2,('加入购物车','2024-01-01 05:25:30',1704057930)) │
│ (2,('查看详情页','2024-01-01 05:45:00',1704059100)) │
│ (2,('查看详情页','2024-01-01 06:15:00',1704060900)) │
│ (2,('查看详情页','2024-01-01 06:25:00',1704061500)) │
│ (2,('查看详情页','2024-01-01 06:27:38',1704061658)) │
│ (2,('查看详情页','2024-01-01 06:35:00',1704062100)) │
│ (2,('用券','2024-01-01 06:40:00',1704062400)) │
│ (3,('登录','2024-01-01 04:12:00',1704053520)) │
│ (3,('领券','2024-01-01 04:42:00',1704055320)) │
│ (3,('收藏商品','2024-01-01 05:15:00',1704057300)) │
│ (3,('加入购物车','2024-01-01 05:25:00',1704057900)) │
│ (3,('用券','2024-01-01 05:40:00',1704058800)) │
└─────────────────────────────────────────────────────┘经过整理之后的数据，看起来就比初始时清晰多了，基本上能够一眼看出哪些用户的行为序列比较长，哪些比较短。
7. 关键行为分析
现在，最关键且最敏感的一步就是要找到用户从领券到用券之间，都做了哪些事，正是这些行为，基本上决定了他是否是一个羊毛党。
:) WITH
-- 将领券事件做标记，如果是领券，那么将其标记为1，否则为0
arrayMap(x -> if(x.2.1 == '领券', 1, 0), sequence) AS masks_start,
-- 按照标记分割用户行为序列
arraySplit((x, y) -> y, sequence, masks_start) AS split_events_start_arr,
-- 将领券之后的行为单独拿出来
split_events_start_arr[2] AS split_events_start,
-- 同理，将用券事件做标记，如果是领券，那么将其标记为1，否则为0
arrayMap(x -> if(x.2.1 == '用券', 1, 0), split_events_start) AS masks_end,
-- 按照标记反向分割用户行为序列
arrayReverseSplit((x, y) -> y, split_events_start, masks_end) AS split_events_end,
-- 将用券之前的行为单独拿出来
split_events_end[1] AS split_events
-- 展开数组
SELECT arrayJoin(split_events)
FROM dws.event_sequence_temporary;
┌─arrayJoin(split_events)─────────────────────────────┐
│ (1,('领券','2024-01-01 05:15:08',1704057308)) │
│ (1,('查看详情页','2024-01-01 05:17:08',1704057428)) │
│ (1,('查看详情页','2024-01-01 05:20:00',1704057600)) │
│ (1,('浏览商品','2024-01-01 05:25:00',1704057900)) │
│ (1,('浏览商品','2024-01-01 05:30:30',1704058230)) │
│ (1,('浏览商品','2024-01-01 05:30:31',1704058231)) │
│ (1,('用券','2024-01-01 05:40:30',1704058830)) │
│ (2,('领券','2024-01-01 04:45:00',1704055500)) │
│ (2,('查看详情页','2024-01-01 05:16:08',1704057368)) │
│ (2,('加入购物车','2024-01-01 05:25:30',1704057930)) │
│ (2,('查看详情页','2024-01-01 05:45:00',1704059100)) │
│ (2,('查看详情页','2024-01-01 06:15:00',1704060900)) │
│ (2,('查看详情页','2024-01-01 06:25:00',1704061500)) │
│ (2,('查看详情页','2024-01-01 06:27:38',1704061658)) │
│ (2,('查看详情页','2024-01-01 06:35:00',1704062100)) │
│ (2,('用券','2024-01-01 06:40:00',1704062400)) │
│ (3,('领券','2024-01-01 04:42:00',1704055320)) │
│ (3,('收藏商品','2024-01-01 05:15:00',1704057300)) │
│ (3,('加入购物车','2024-01-01 05:25:00',1704057900)) │
│ (3,('用券','2024-01-01 05:40:00',1704058800)) │
└─────────────────────────────────────────────────────┘按照以上的这些操作，就把与风险行为相关的事件全部提取出来了，而且按照需要做了排列，这为后续业务处理提供了充分的依据。
实际的大数据风控远比这复杂，而且也可能会用到更多更复杂的函数，例如开窗函数和聚合函数的组合。
但万变不离其宗，再怎么复杂的分析和查询，都是由这些基本的SQL语句、函数、数据类型和库表引擎拼装起来的。
感谢支持
更多内容，请移步《超级个体》。