处理亿级别数据,这次没翻车

❝

我也没想到我写的代码怎么处理了上亿的数据量，重点是它没有翻车. 我预估的数据处理量应该在千万级别,运行了几天后我再去查看发现处理了1亿的数据量. 无论出于什么原因,它依然运行良好,总而言之 It’s work !

❞

设计

一开始产品说让我们做一个销售报表把,帮助运营同学更好的了解销售情况,调整销售策略提高销量。在深入了解了需求后我们说不如我们来搞个数据仓库吧,于是我们开始着手构建数据仓库，以便更好地管理和分析销售数据。开始我做了一个简陋的设计, 非常的简陋是吧,依照这个设计图纸去施工肯定是不现实的。所以我需要对它进行细化。

通过这个初始的简陋设计依然可以从宏观抽象层面体现出要做一个数仓的大致步骤:

数据的收集
数据的整形
数据的存储
如何对外提供数据能力

细化设计

在这个架构中我们会去监听 MySQL 的 binlog 日志, 这样做不会对业务产生任何的侵入,而是把我们的关注点从业务层面剥离开,专注于数据本身。

在对 Binlog Listening Service 组件的实现过程中,考虑过一些开源产品比如 Canal (主要用途是基于 MySQL 数据库的增量日志 Binlog 解析，提供增量数据订阅和消费。).

但是出于部署成本,学习成本等因素考虑,最终还是没有用,而是基于 mysql-binlog-connector-java 自己写了一个。有同学可能就要问了,已经有开源成熟的产品了为什么还要自己写呢？这不是重复造轮子么? 没错,就是玩…… , 其实开发的过程还是挺有趣的,解决了一些以前没遇到的问题,对 MySQL binlog 的一些知识和细节掌握的更多更深入了。基本原理也是将服务做为 MySQL 的一个 slave 然后进行数据同步。其中要考虑到数据的顺序性问题,以及如何将拿到的 binlog 行数据匹配到对应表具体的列上。在大量行变更的情况下会导致 JVM 的频繁 GC 等问题。

<dependency>
    <groupId>com.github.shyiko</groupId>
    <artifactId>mysql-binlog-connector-java</artifactId>
    <version>0.21.0</version>
</dependency>

将封装好的数据发送到 kafka 集群,由消费者 Data Handler 进行数据整形.整形后的数据会被交给 Data Collector (数据收集器组件)。由 Data Collector 负责将数据存储到数据仓库中。Data Collector 可以支持任意的数据存储类型,无论是 Starrocks 还是 MySQL或者是其他的数据存储产品,扩展起来非常的容易。

在投递 kafka 时会给每个消息携带一个时间序列相关的唯一消息 id , 每个表会分为 table_Update Topic 和 table_Insert_And_Delete Topic, 避免消息在同一个队列上挤压堆积,导致数据同步的实时性不高。在数据同步时会根据这个携带的唯一 id 进行乐观锁更新,也就是说如果旧的变更后执行了并不会对数据的正确性造成影响。

我们的其他业务服务需要从数仓中获取数据,在业务服务中本身连接了自己的业务数据库,同时要从数仓中获取数据,这时候会通过 SQL Routing 组件将 SQL 路由到数仓中去执行,不对业务系统的数据访问造成任何影响和侵入。SQL Routing 也是我自己造的轮子,别问我为什么就是玩…