中国好胖子手摸手带你搭建实时数仓，一顿烧烤就行！
实时数仓应用场景
场景决定一切。离线数仓的时候数据更新频率是T+1，也就是说必须隔一天才能看到结果，今天看昨天的数据。但是数据界有一个确定的结论，就是数据越新，价值越大。于是就有了推荐、风控等各种实时应用场景，让数据在最有价值的时候被利用好。在这些场景中，对数据的实时性要求就非常高，往往需要毫秒级反应，否则会影响用户体验，带来不必要的损失。
在最开始的时候，业界采用Storm进行实时数据流计算。后来有了spark streaming，现在最火热的当属Flink了。在离线数据仓库架构设计的时候，大家知道需要分层，数据得落地在数据存储介质中，一般是各种数据库。但是实时场景，数据一直是在流动的，数据怎么落地？怎么分层？以下图为例，数据从各种日志中实时读取过来，最后流向实时大屏，大屏计算结果就必须得有个地方存着啊。
实时数仓设计中需要考虑的问题
上图看上去很不错，能在大屏上直接展示结果。但是一细看，就会有无数问题：大屏上需要展示多少指标？面对任性的业务，面对他们 无穷无尽 的需求，作为技术能做的是怎么能更好的服务他们？如何做到以数据驱动业务的成长，以数据驱动产业数字化？
业务的需求多变，指标可能是无穷无尽的，导致的也就是开发速度可能不尽人意。可能两天才有一个指标的产出，复杂的可能一个星期乃至更长。如果需求不能加以控制，我们将陷入无尽的任务中。如果拒绝需求，业务的需求得不到满足，数据团队存在的意义又会大大降低。我们该怎么办？
那么我们有没有可能在牺牲一些查询速度的同时，来提升我们的开发速度，我们应该都知道spark streaming 和flink都是支持sql开发的。那么flink 或者spark streaming 来进行sql 开发，时效性和灵活性会比较低，直接开放给业务方，用户体验会非常不好。这是一个很值得思考的问题。那么我们又该咋办？
我们是不是可以尝试将我们的binlog 数据以及埋点数据进行拉宽，也就是宽表化的一些操作？变成离线数仓那样的OLAP？自主化的查询呢？对，这就是实时数仓的诞生！
实时数仓在我理解中呢，可以对外进行服务，并且可以实时的进行OLAP查询，也就是在线化查询 Ad-hoc化的查询。
实战中实时数仓的架构演进
初始
在我刚来公司的时候，并没有实时数仓，只是一些批处理化的操作，并且是一种烟囱式的开发。数据流程是这样的，我们采用的greenplum来做的准实时数仓，每15分钟去业务库和神策系统拉取实时数据到greenplum中进行计算，如下图所示。
我们这个时候肯定可以发现，假如指标多的时候，那么对于开发速度来讲是一个十分缓慢的一个过程，并且会造成很多数据的冗余计算，有些指标并不能复用。
实时数仓0.1
在对公司情况有所了解之后，我选择了Flink作为实时数仓的核心组件。在熟悉了业务之后，我选了一个线上分析的需求，简单梳理了一下数据流向：
上线之后效果还不错，业务方非常满意，领导也加以赞许。但是后来慢慢的需求多了起来，我意识到拿flink写需求肯定会陷入无尽的任务。我明白必须要避免烟囱式的开发，应该做好数据架构，把数据和业务彻底解耦！
实时数仓1.0
其实实时数仓也很简单，你把实时表都想象成离线宽表一样的话，那么直接在宽表上进行计算不就好了吗。OK，咱开始实战。
按照CIF设计规范，要拉宽这些数据表，得有核心场景。我们公司是比较关心销售的，把离线的销售宽表拿实时展现出来就是很好的一个场景。
先确定数据源，我们的数据是从业务库来。但是我们的postgresql 比较老，并没有binlog这些的操作。我当初就和研发的架构探讨了一下，他们那边借助触发器来进行给我往kafka中打数据，解决数据源的问题。
然后解决数据质量问题，我对数据先进行了校验，也就是看看我要的字段是否都齐全，数据是否有问题。这两个问题都解决之后，就开始尝试用flink接入kafka的数据。
这个时候我数据是拿到了，但是我需要拉宽，我应该怎么拉宽？我选择把相关维度表放置到redis，这样比较快。这样在flink的map方法中进行查询redis中的数据来进行拉宽维度表。
这个时候就来问题了，我的维度表是会更新的啊。我也就问了我们组的业务大佬，咨询了一下，发现维度表无非就是门店维表、品类维表（一级类、二级类、三级类等）、城市维表、商品维表、主推表等。这些都是缓慢变化维，即使是更新，也会提前上线几天进行更新。并且我们在凌晨0点到2点是不进行出库操作的，我在这个时候进行维度表更新操作不就好了吗。那么也就有下面的设计，定时进行redis中的维度数据更新：
那么接下来我们就可以进行销售宽表的拉宽操作了，但是我这个时候又发现了一个问题，我拉宽之后存在哪里，这个时候我得思考的几点是。第一单表查询足够快、最好支持join。那么我开始的调研过几款 Tidb、Doris、Druid、Clickhouse。我在单机测试的表现上来看，clickhouse给我带来了无与伦比的感觉。并且考虑到当时的业务场景，也就毅然决然的采用了Clickhouse为基础的实时数仓。
实时数仓2.0
然后就这样的一个架构持续了大概两个月的时间，业务也越来越复杂。比如：门店需要对导购拉新来做当日的绩效考核，因此需要接入一些用户维度表。那么我们总计有2000多万的用户，我全部都导入的redis话会有一些问题。同时还有一些实时需求，例如要在用户宽表中标记出来这个用户是不是新会员、是否是孕妇等。
另外还有一些交易回溯分摊的问题，例如一个用户购买了一个A 产品，赠了一个B产品，那么这个时候，品类间的毛利就有了一些损失。例如买一件衣服送一罐奶粉，那么这个时候就有了问题，不同品类的负责人不干了，因为赠品的KPI少了。衣服的总监愿意啊，买的人多了，但是奶粉的总监不干了，我毛利没了啊。所以就有了回溯分摊这一个事情。人生太艰难了，解决技术问题，还得解决业务问题。
。
我就写了个flink程序，自定义了一个source实时的去库里面拉取数据，因为没有binlog。但是不能实时的去啊，对库的影响太大了。那么这个时候就想到了我每次间隔一分钟去拉取一次放到redis当中，然后flink join 的时候就写入到clickhouse中。
对于没有join上的，就放到kafka的另一个topic中例如 dws_sold_detail_retry 然后再开一个flink 专门消费这个。假如还没有join上 就继续放到这个topic中，在日志中追加一个重试次数，假如这个消息重试了超过5次，则认为消费失败，不再消费。为了避免此类情况影响统计结果，我增加了一个实时数据监控，每天的销售额差异不能超过百分之3，超过就报警，进行人工干预。
就这样，慢慢的加入了其他的一些宽表，例如库存、优惠券、会员宽表、促销宽表等。但是慢慢的问题也有了，那就是flink写入clickhouse的时候假如表特别宽的话，代码量是很大的。后来我就引入了waterdrop。
也就是以上的架构图。
实战总结与发展
在以上的架构中，我的核心思想就是，用flink拉宽、计算之后，交给olap引擎做多维分析，对数据和业务进行解耦。这个架构是灵活可扩展的，部分组件是可以完美的可插拔的，例如flink可以改成spark streaming、storm。clickhouse可以根据不同的业务场景更改为tidb、drois、greenplum、kudu等。
那么上面的架构也有一些问题，例如维度表太大了怎么办，后面我又引入了二级缓存。也就是引入的hbase，并且支持对外提供查询三个月内的数据实时查询。
最终架构图：
