基于Flink+ClickHouse打造轻量级点击流实时数仓

Flink 和 ClickHouse 分别是实时计算和&＃xff08;近实时&＃xff09;OLAP 领域的翘楚&＃xff0c;也是近些年非常火爆的开源框架&＃xff0c;很多大厂都在将两者结合使用来构建各种用途的实时平台&＃xff0c;效果很好。关于两者的优点就不再赘述&＃xff0c;本文来简单介绍笔者团队在点击流实时数仓方面的一点实践经验。

点击流及其维度建模

所谓点击流&＃xff08;click stream&＃xff09;&＃xff0c;就是指用户访问网站、App 等 Web 前端时在后端留下的轨迹数据&＃xff0c;也是流量分析&＃xff08;traffic analysis&＃xff09;和用户行为分析&＃xff08;user behavior analysis&＃xff09;的基础。点击流数据一般以访问日志和埋点日志的形式存储&＃xff0c;其特点是量大、维度丰富。以我们一个中等体量的普通电商平台为例&＃xff0c;每天产生约 200GB 左右、数十亿条的原始日志&＃xff0c;埋点事件 100&＃43; 个&＃xff0c;涉及 50&＃43; 个维度。

按照 Kimball 的维度建模理论&＃xff0c;点击流数仓遵循典型的星形模型&＃xff0c;简图如下。

点击流数仓分层设计

点击流实时数仓的分层设计仍然可以借鉴传统数仓的方案&＃xff0c;以扁平为上策&＃xff0c;尽量减少数据传输中途的延迟。简图如下。

• DIM 层&＃xff1a;维度层&＃xff0c;MySQL 镜像库&＃xff0c;存储所有维度数据。
• ODS 层&＃xff1a;贴源层&＃xff0c;原始数据由 Flume 直接进入 Kafka 的对应 topic。
• DWD 层&＃xff1a;明细层&＃xff0c;通过 Flink 将 Kafka 中数据进行必要的 ETL 与实时维度 join 操作&＃xff0c;形成规范的明细数据&＃xff0c;并写回 Kafka 以便下游与其他业务使用。再通过 Flink 将明细数据分别写入 ClickHouse 和 Hive 打成大宽表&＃xff0c;前者作为查询与分析的核心&＃xff0c;后者作为备份和数据质量保证&＃xff08;对数、补数等&＃xff09;。
• DWS 层&＃xff1a;服务层&＃xff0c;部分指标通过 Flink 实时汇总至 Redis&＃xff0c;供大屏类业务使用。更多的指标则通过 ClickHouse 物化视图等机制周期性汇总&＃xff0c;形成报表与页面热力图。特别地&＃xff0c;部分明细数据也在此层开放&＃xff0c;方便高级 BI 人员进行漏斗、留存、用户路径等灵活的 ad-hoc 查询&＃xff0c;这些也是 ClickHouse 远超过其他 OLAP 引擎的强大之处。

要点与注意事项

Flink 实时维度关联

Flink 框架的异步 I/O 机制为用户在流式作业中访问外部存储提供了很大的便利。针对我们的情况&＃xff0c;有以下三点需要注意&＃xff1a;

• 使用异步 MySQL 客户端&＃xff0c;如 Vert.x MySQL Client。
• AsyncFunction 内添加内存缓存&＃xff08;如 Guava Cache、Caffeine 等&＃xff09;&＃xff0c;并设定合理的缓存驱逐机制&＃xff0c;避免频繁请求 MySQL 库。
• 实时维度关联仅适用于缓慢变化维度&＃xff0c;如地理位置信息、商品及分类信息等。快速变化维度&＃xff08;如用户信息&＃xff09;则不太适合打进宽表&＃xff0c;我们采用 MySQL 表引擎将快变维度表直接映射到 ClickHouse 中&＃xff0c;而 ClickHouse 支持异构查询&＃xff0c;也能够支撑规模较小的维表 join 场景。未来则考虑使用 MaterializedMySQL 引擎&＃xff08;当前仍未正式发布&＃xff09;将部分维度表通过 binlog 镜像到 ClickHouse。

Flink-ClickHouse Sink 设计

可以通过 JDBC&＃xff08;flink-connector-jdbc&＃xff09;方式来直接写入 ClickHouse&＃xff0c;但灵活性欠佳。好在 clickhouse-jdbc 项目提供了适配 ClickHouse 集群的 BalancedClickhouseDataSource 组件&＃xff0c;我们基于它设计了 Flink-ClickHouse Sink&＃xff0c;要点有三&＃xff1a;

• 写入本地表&＃xff0c;而非分布式表&＃xff0c;老生常谈了。
• 按数据批次大小以及批次间隔两个条件控制写入频率&＃xff0c;在 part merge 压力和数据实时性两方面取得平衡。目前我们采用 10000 条的批次大小与 15 秒的间隔&＃xff0c;只要满足其一则触发写入。
• BalancedClickhouseDataSource 通过随机路由保证了各 ClickHouse 实例的负载均衡&＃xff0c;但是只是通过周期性 ping 来探活&＃xff0c;并屏蔽掉当前不能访问的实例&＃xff0c;而没有故障转移——亦即一旦试图写入已经失败的节点&＃xff0c;就会丢失数据。为此我们设计了重试机制&＃xff0c;重试次数和间隔均可配置&＃xff0c;如果当重试机会耗尽后仍然无法成功写入&＃xff0c;就将该批次数据转存至配置好的路径下&＃xff0c;并报警要求及时检查与回填。

当前我们仅实现了 DataStream API 风格的 Flink-ClickHouse Sink&＃xff0c;随着 Flink 作业 SQL 化的大潮&＃xff0c;在未来还计划实现 SQL 风格的 ClickHouse Sink&＃xff0c;打磨健壮后会适时回馈给社区。另外&＃xff0c;除了随机路由&＃xff0c;我们也计划加入轮询和 sharding key hash 等更灵活的路由方式。

还有一点就是&＃xff0c;ClickHouse 并不支持事务&＃xff0c;所以也不必费心考虑 2PC Sink 等保证 exactly once 语义的操作。如果 Flink 到 ClickHouse 的链路出现问题导致作业重启&＃xff0c;作业会直接从最新的位点&＃xff08;即 Kafka 的 latest offset&＃xff09;开始消费&＃xff0c;丢失的数据再经由 Hive 进行回填即可。

ClickHouse 数据重平衡

ClickHouse 集群扩容之后&＃xff0c;数据的重平衡&＃xff08;reshard&＃xff09;是一件麻烦事&＃xff0c;因为不存在类似 HDFS Balancer 这种开箱即用的工具。一种比较简单粗暴的思路是修改 ClickHouse 配置文件中的 shard weight&＃xff0c;使新加入的 shard 多写入数据&＃xff0c;直到所有节点近似平衡之后再调整回来。但是这会造成明显的热点问题&＃xff0c;并且仅对直接写入分布式表才有效&＃xff0c;并不可取。

因此&＃xff0c;我们采用了一种比较曲折的方法&＃xff1a;将原表重命名&＃xff0c;在所有节点上建立与原表 schema 相同的新表&＃xff0c;将实时数据写入新表&＃xff0c;同时用 clickhouse-copier 工具将历史数据整体迁移到新表上来&＃xff0c;再删除原表。当然在迁移期间&＃xff0c;被重平衡的表是无法提供服务的&＃xff0c;仍然不那么优雅。如果大佬们有更好的方案&＃xff0c;欢迎交流。

作者&＃xff1a;LittleMagic

原文链接

本文为阿里云原创内容&＃xff0c;未经允许不得转载。