0x00 TLDR;
AWS S3中存储了大量日志,按年月日的目录进行存储,日志中每行放一个json字符串。
我们的目标是,以日志文件json格式为参考,在clickhouse中创建相应表构建
然后通过clickhouse提供的能力把,日志文件内容导入 clickhouse系统。
0x01 ClickHouse集群模式简介
clickhouse 集群 和
cluster
是两个概念。
集群
,可能泛指由多个clickhouse实例组成的集合。
cluster
可以理解为数据存储逻辑 ,根据配置可以使用整个集群的部分实例。
假设我们已经提前部署好4节点clickhouse集群,现在我们来配置一个 cluster
根据自己的数据重要等级,这里使用4个分片,0备份。这样可以在有限的磁盘上,放更多的数据。
在每个clickhouse节点中增加配置:
vim /etc/clickhouse-server/metrika.xml
<yandex>
<clickhouse_remote_servers>
<log_cluster>
<shard>
<replica>
<host>10.10.0.1</host>
<port>9000</port>
<user>default</user>
<password>yourpwd</password>
</replica>
</shard>
<shard>
<replica>
<host>10.10.0.2</host>
<port>9000</port>
<user>default</user>
<password>yourpwd</password>
</replica>
</shard>
<shard>
<replica>
<host>10.10.0.3</host>
<port>9000</port>
<user>default</user>
<password>yourpwd</password>
</replica>
</shard>
<shard>
<replica>
<host>10.10.0.4</host>
<port>9000</port>
<user>default</user>
<password>yourpwd</password>
</replica>
</shard>
</log_cluster>
<!-- 下面是其他默认配置, zookeeper 等 -->
每个节点clickhouse 服务重启生效后,可以查看 cluster信息:
select * from system.clusters;
0x02 创建 database 和 table
和mysql一样,clickhouse提供了相同的概念:database, table。
在上面的cluster 之上,先来创建一个database:
create database if not exists my_database on cluster log_cluster;
接下来,讨论一下集群模式下表的类型。
ReplicatedMergeTree
写数据时,为了方便后面查询,会根据特定的key,把数据写到不同的 分片中。Distributed
读数据时,因为数据是在多个分片中,需要一个逻辑层进行聚合。
在建表前,我们假定单行 json 格式:
{
"event": "click",
"eventParams": {
"app_id": "123456",
"time": "1721174974991",
}
}
接下来建一个写入表, 要求把多层级的json数据,放到同一个维度里。
create table if not exists my_database.my_table on cluster log_cluster
(
event_time DateTime,
event String,
event_param_app_id String,
) engine = ReplicatedMergeTree('/clickhouse/tables/{layer}-{shard}/mydatabase/mytable', '{replica}')
partition by toYYYYMMDD(event_time)
order by (event, event_param_app_id, event_time);
上面使用的ReplicatedMergeTree表引擎,定义了在zookeeper中的参数。
然后再定义一个只读表:
create table if not exists my_database.my_table_ro on cluster log_cluster
(
event_time DateTime,
event String,
event_param_app_id String,
) engine = Distributed(log_cluster, my_database, my_table);
只读表的Distributed表引擎参数,比较好理解,见名知义。
0x03 导入SQL
clickhouse 对三方支持较强,这里使用的是 s3 function
功能,直接在sql中读取 s3文件内容。
假定已提前配置好 s3 的iam权限,准备好相应的 ak,sk.
如下面的SQL示例,有几个细节:
- 时间格式转换,由时间戳转内部的DateTime
- S3文件路径的通配符
- JSON多层级对象的提取
经测试发现:JSONEachRow返回的 raw_data 对象,子层级都是以String的方式使用。
在业务设计时,尽量避免多层级的数据格式。
INSERT INTO my_database.my_table (event_time,
event,
event_param_app_id)
SELECT toDateTime(JSONExtractUInt(raw_data.eventParams, 'time') / 1000) AS event_time,
raw_data.event AS event,
JSONExtractString(raw_data.eventParams, 'app_id') AS event_param_app_id,
FROM s3(
'https://your_bucket.s3.eu-central-1.amazonaws.com/2024/01/*/*/*',
'your_s3_ak',
'your_s3_sk',
'JSONEachRow') as raw_data;
查询结果:
select count(1) from my_database.my_table_ro;
0x04 工程化&QA
上述SQL如果数据量较大时,在普通的SQL终端执行可能会发现超时问题。
可使用 clickhouse-client 客户端方式访问。
单分区单次insert 默认配置为100, 可以在会话级别设置上限,避免提交失败。
set max_partitions_per_insert_block=10000;
最后,如何每日自动同步数据呢,可以crontab + shell 方式:
# 写的同步SQL,
# 在命令行调用 clickhouse-client 执行sql
clickhouse-client -h 127.0.0.1 --port 9000 -u default --password $yourpwd --query "$sql"
0x05 创建只读用户
因为 cluster 模式下,需要把账号和赋权信息都要同步到各个节点。 所以,在创建用户和grant语句中,都有 on cluster
子句
-- 创建用户
create user if not exists readonly_user on cluster log_cluster identified with plaintext_password by '123456';
-- 只读权限
GRANT ON CLUSTER log_cluster SHOW TABLES, SELECT ON my_database.* TO readonly_user with grant option;
-- 查看用户当前权限
show grants for readonly_user;
0x06 高效删表
-- 查询表的分片
SELECT partition
FROM system.parts
WHERE table = 'user_data';
-- 20240905
-- 20240913
-- 20240920
-- 20240921
-- 20240922
-- 删除指定分片
ALTER TABLE your_databse.user_data on cluster your_cluster
DROP PARTITION '20240905';
select count(*)
from your_databse.user_data_ro;