实时数仓模型

为了计算一些实时指标，就在原来离线数仓的基础上增加了一个实时计算的链路，并对数据源做流式改造（即把数据发送到消息队列），实时计算去订阅消息队列，直接完成指标增量的计算，推送到下游的数据服务中去，由数据服务层完成离线&实时结果的合并。

实时数仓主要是基于数据采集工具，如canal等原始数据写入到kafka这样的数据通道中，最后一般都是写入到类似于HBase这样的OLAP存储系统中。对外提供分钟级别，甚至秒级别的查询方案。

问题：

同样的需求需要开发两套一样的代码：这是 Lambda 架构最大的问题，两套代码不仅仅意味着开发困难（同样的需求，一个在批处理引擎上实现，一个在流处理引擎上实现，还要分别构造数据测试保证两者结果一致），后期维护更加困难，比如需求变更后需要分别更改两套代码，独立测试结果，且两个作业需要同步上线。

资源占用增多：同样的逻辑计算两次，整体资源占用会增多（多出实时计算这部分）

Kappa 架构：

Lambda 架构虽然满足了实时的需求，但带来了更多的开发与运维工作，其架构背景是流处理引擎还不完善，流处理的结果只作为临时的、近似的值提供参考。后来随着 Flink 等流处理引擎的出现，流处理技术很成熟了，这时为了解决两套代码的问题，LickedIn 的 Jay Kreps 提出了 Kappa 架构。

问题：

Kappa 架构可以认为是 Lambda 架构的简化版（只要移除 lambda 架构中的批处理部分即可）。

在 Kappa 架构中，需求修改或历史数据重新处理都通过上游重放完成。

Kappa 架构最大的问题是流式重新处理历史的吞吐能力会低于批处理，但这个可以通过增加计算资源来弥补。

Kappa 架构可能也需要利用离线的数据进行校验。

因为我们经常会面临业务变更，所以很多业务逻辑是需要去迭代的。之前产出的一些数据，如果口径变更了，就需要重算，甚至重刷历史数据。对于实时数仓来说，怎么去解决数据重算问题？

Kappa 架构在这一块的思路是：首先要准备好一个能够存储历史数据的消息队列，比如 Kafka，并且这个消息对列是可以支持你从某个历史的节点重新开始消费的。 接着需要新起一个任务，从原来比较早的一个时间节点去消费 Kafka 上的数据，然后当这个新的任务运行的进度已经能够和现在的正在跑的任务齐平的时候，你就可以把现在任务的下游切换到新的任务上面，旧的任务就可以停掉，并且原来产出的结果表也可以被删掉。

Lambda 架构与 Kappa 架构的对比

在真实的场景中，很多时候并不是完全规范的 Lambda 架构或 Kappa 架构，可以是两者的混合，比如大部分实时指标使用 Kappa 架构完成计算，少量关键指标（比如金额相关）使用 Lambda 架构用批处理重新计算，增加一次校对过程。

Kappa 架构并不是中间结果完全不落地，现在很多大数据系统都需要支持机器学习（离线训练），所以实时中间结果需要落地对应的存储引擎供机器学习使用，另外有时候还需要对明细数据查询，这种场景也需要把实时明细层写出到对应的引擎中。

实时数仓模型

实时数仓需要解决的问题：

1）第一，要支持同时读写，就意味着你写的时候还可以读，不应该读到一个错误的结果。同时还可以支持多个写，且能保证数据的一致性；

2）第二，可以高吞吐地从大表读取数据。大数据方案不能有诸多限制，比如，我听说有些方案里最多只可以支持几个并发读，或者读的文件太多了就不让你提交作业了。如果这样，对业务方来说，你的整个设计是不满足他的需求的；

3）第三，错误是无可避免，你要可以支持回滚，可以重做，或者可以删改这个结果，不能为了支持删改而要求业务方去做业务逻辑的调整；

4）第四，在重新改变业务逻辑的时候要对数据做重新处理，这个时候，业务是不能下线的。在数据被重新处理完成之前，数据湖的数据是要一直可被访问的；

5）第五，因为有诸多原因，数据可能会有晚到的情况，你要能处理迟到数据而不推迟下阶段的数据处理。

实时数仓的实施关键点：

端到端数据延迟、数据流量的监控

故障的快速恢复能力

数据的回溯处理，系统支持消费指定时间段内的数据

实时数据从实时数仓中查询，T+1数据借助离线通道修正

数据地图、数据血缘关系的梳理

业务数据质量的实时监控，初期可以根据规则的方式来识别质量状况

ODS 层的建设

1.首先就是数据源的来源尽可能要统一 。

第一个统一就是实时的 数据源本身要跟自己统一 

第二个统一是指 实时和离线的统一 ，这个统一可能更重要一点

2.第二个要点就是数据乱序的问题，

我们在采集数据的时候会有一个比较大的问题，可能同一条数据，由于分区的存在，这条数据先发生的状态后消费到，后发生的状态先消费到。我们在解决这一问题的时候采用的是美团内部的一个数据组件。

DW 层的建设

明细层的建设思路其实跟离线数仓的基本一致，主要在于如何解决 ODS 层的数据可能存在的数据噪声、不完整和形式不统一的问题，让它在仓库内是一套满足规范的统一的数据源。我们的建议是如果有可能的话，最好入什么仓怎么入仓，这个过程和离线保持一致。

1.数据解析

2.业务整合

3.脏数据清洗

4.模型规范化

重复数据处理

除了数据本身我们会在每条数据上额外补充一些信息，应对实时数据生产环节的一些常见问题

唯一键和主键

我们会给每一条数据都补充一个唯一键和一个主键，这两个是一对的，唯一键就是标识是唯一一条数据的，主键是标记为一行数据。一行数据可能变化很多次，但是主键是一样的，每一次变化都是其一次唯一的变化，所以会有一个唯一键。唯一键主要解决的是数据重复问题，从分层来讲，数据是从我们仓库以外进行生产的，所以很难保证我们仓库以外的数据是不会重复的。

可能有些人交付数据给也会告知数据可能会有重复。生成唯一键的意思是指我们需要保证 DW 层的数据能够有一个标识，来解决可能由于上游产生的重复数据导致的计算重复问题。生成主键，其实最主要在于主键在 kafka 进行分区操作，跟之前接 ODS 保证分区有序的原理是一样的，通过主键，在 kafka 里进行分区之后，消费数据的时候就可以保证单条数据的消费是有序的。

版本和批次

版本和批次这两个其实又是一组。当然这个内容名字可以随便起，最重要的是它的逻辑。

首先，版本。版本的概念就是对应的表结构，也就是 schema 一个版本的数据。由于在处理实时数据的时候，下游的脚本依赖表上一次的 schema 进行开发的。当数据表结构发生变化的时候，就可能出现两种情况：第一种情况，可能新加或者删减的字段并没有用到，其实完全不用感知，不用做任何操作就可以了。另外一种情况， 需要用到变动的字段。此时会产生一个问题，在 Kafka 的表中，就相当于有两种不同的表结构的数据。这时候其实需要一个标记版本的内容来告诉我们，消费的这条数据到底应该用什么样的表结构来进行处理，所以要加一个像版本这样的概念。

第二，批次。批次实际上是一个更不常见的场景，有些时候可能会发生数据重导，它跟重启不太一样，重启作业可能就是改一改，然后接着上一次消费的位置启动。而重导的话，数据消费的位置会发生变化。

比如，今天的数据算错了，领导很着急让我改，然后我需要把今天的数据重算，可能把数据程序修改好之后，还要设定程序，比如从今天的凌晨开始重新跑。这个时候由于整个数据程序是一个 7x24 小时的在线状态，其实原先的数据程序不能停，等重导的程序追上新的数据之后，才能把原来的程序停掉，最后使用重导的数据来更新结果层的数据。

在这种情况下，必然会短暂的存在两套数据。这两套数据想要进行区分的时候，就要通过批次来区分。其实就是所有的作业只消费指定批次的数据，当重导作业产生的时候，只有消费重导批次的作业才会消费这些重导的数据，然后数据追上之后，只要把原来批次的作业都停掉就可以了，这样就可以解决一个数据重导的问题。

■ 维度数据建设

其次就是维度数据，我们的明细层里面包括了维度数据。关于维度的数据的处理，实际上是先把维度数据分成了两大类采用不同的方案来进行处理。

变化频率低的维度

第一类数据就是一些变化频率比较低的数据，这些数据其实可能是一些基本上是不会变的数据。比如说，一些地理的维度信息、节假日信息和一些固定代码的转换。

这些数据实际上我们采用的方法就是直接可以通过离线仓库里面会有对应的维表，然后通过一个同步作业把它加载到缓存中来进行访问。还有一些维度数据创建得会很快，可能会不断有新的数据创建出来，但是一旦创建出来，其实也就不再会变了。

比如说，美团上开了一家新的门店，门店所在的城市名字等这些固定的属性，其实可能很长时间都不会变，取最新的那一条数据就可以了。这种情况下，我们会通过公司内部的一些公共服务，直接去访问当前最新的数据。最终，我们会包一个维度服务的这样一个概念来对用户进行屏蔽，具体是从哪里查询相关细节，通过维度服务即可关联具体的维度信息。

变化频率高的维度

第二类是一些变化频率较高的数据。比如常见的病人心脑科的状态变动，或者某一个商品的价格等。这些东西往往是会随着时间变化比较频繁，比较快。而对于这类数据，我们的处理方案就稍微复杂一点。首先对于像价格这样变化比较频繁的这种维度数据，会监听它的变化。比如说，把价格想象成维度，我们会监听维度价格变化的消息，然后构建一张价格变换的拉链表。

一旦建立了维度拉链表，当一条数据来的时候，就可以知道，在这个数据某一时刻对应的准确的维度是多少，避免了由于维度快速的变化导致关联错维度的问题。

另一类如新老客这维度，于我们而言其实是一种衍生维度，因为它本身并不是维度的计算方式，是用该用户是否下过单来计算出来的，所以它其实是用订单数据来算出来的一个维度。

所以类似订单数的维度，我们会在 DW 层建立一些衍生维度的计算模型，然后这些计算模型输出的其实也是拉链表，记录下一个用户每天这种新老客的变化程度，或者可能是一个优质用户的变化的过程。由于建立拉链表本身也要关联维度，所以可以通过之前分组 key 的方式来保障不乱序，这样还是将其当做一个不变的维度来进行关联。

通过这种方式来建立拉链表相对麻烦，所以实际上建议利用一些外部组件的功能。实际操作的时候，我们使用的是 Hbase。HBase 本身支持数据多版本的，而且它能记录数据更新的时间戳，取数据的时候，甚至可以用这个时间戳来做索引。

所以实际上只要把数据存到 HBase 里，再配合上 mini-versions ，就可以保证数据不会超时死掉。上面也提到过，整个实时数仓有一个大原则，不处理离线数仓能处理的过程。相当于处理的过程，只需要处理三天以内的数据，所以还可以通过配置 TTL 来保证 HBase 里的这些维度可以尽早的被淘汰掉。因为很多天以前的维度，实际上也不会再关联了，这样就保证维度数据不会无限制的增长，导致存储爆炸。

■ 维度数据使用

处理维度数据之后，这个维度数据怎么用？

第一种方案，也是最简单的方案，就是使用 UDTF 关联。其实就是写一个 UDTF 去查询上面提到的维度服务，具体来讲就是用 LATERAL TABLE 关键词来进行关联，内外关联都是支持的。

另外一种方案就是通过解析 SQL ，识别出关联的维表以及维表中的字段，把它原本的查询进行一次转化为原表.flatmap (维表)，最后把整个操作的结果转换成一张新的表来完成关联操作。

但是这个操作要求使用者有很多周边的系统来进行配合，首先需要能解析 SQL ，同时还能识别文本，记住所有维表的信息，最后还要可以执行 SQL 转化，所以这套方案适合一些已经有成熟的基于 Flink SQL 的 SQL开发框架的系统来使用。如果只是单纯的写封装的代码，建议还是使用 UDTF 的方式来进行关联会非常的简单，而且效果也是一样的。

■ 汇总层的建设

在建设实时数仓的汇总层的时候，跟离线的方案其实会有很多一样的地方。

第一点是对于一些共性指标的加工，比如说 pv、uv、交易额这些运算，我们会在汇总层进行统一的运算。另外，在各个脚本中多次运算，不仅浪费算力，同时也有可能会算错，需要确保关于指标的口径是统一在一个固定的模型里面的。本身 Flink SQL 已经其实支持了非常多的计算方法，包括这些 count distinct 等都支持。

值得注意的一点是，它在使用 count distinct 的时候，他会默认把所有的要去重的数据存在一个 state 里面，所以当去重的基数比较大的时候，可能会吃掉非常多的内存，导致程序崩溃。这个时候其实是可以考虑使用 一些非精确系统的算法，比如说 BloomFilter 非精确去重、 HyperLogLog 超低内存去重方案，这些方案可以极大的减少内存的使用。

第二点就是 Flink 比较有特色的一个点，就是 Flink 内置非常多的这种时间窗口。Flink SQL 里面有翻滚窗口、滑动窗口以及会话窗口，这些窗口在写离线 SQL 的时候是很难写出来的，所以可以开发出一些更加专注的模型，甚至可以使用一些在离线开发当中比较少使用的一些比较小的时间窗口。

比如说，计算最近10分钟的数据，这样的窗口可以帮助我们建设一些基于时间趋势图的应用。但是这里面要注意一点，就是一旦使用了这个时间窗口，要配置对应的 TTL 参数，这样可以减少内存的使用，提高程序的运行效率。另外，如果 TTL 不够满足窗口的话，也有可能会导致数据计算的错误。

第三点，在汇总层进行多维的主题汇总，因为实时仓库本身是面向主题的，可能每一个主题会关心的维度都不一样，所以我们会在不同的主题下，按照这个主题关心的维度对数据进行一些汇总，最后来算之前说过的那些汇总指标。但是这里有一个问题，如果不使用时间窗口的话，直接使用 group by ，它会导致生产出来的数据是一个 retract 流，默认的 kafka 的 sink 它是只支持 append 模式，所以在这里要进行一个转化。

如果想把这个数据写入 kafka 的话，需要做一次转化，一般的转化方案实际上是把撤回流里的 false 的过程去掉，把 true 的过程保存起来，转化成一个 append stream ，然后就可以写入到 kafka 里了。

第四点，在汇总层会做一个比较重要的工作，就是衍生维度的加工。如果衍生维度加工的时候可以利用 HBase 存储，HBase 的版本机制可以帮助你更加轻松地来构建一个这种衍生维度的拉链表，可以帮助你准确的 get 到一个实时数据当时的准确的维度。

仓库质量保证

经过上面的环节，如果你已经建立好了一个仓库，你会发现想保证仓库的正常的运行或者是保证它高质量的运行，其实是一个非常麻烦的过程，它要比一线的操作复杂得多，所以我们在建设完仓库之后，需要建设很多的周边系统来提高我们的生产效率。

下面介绍一下我们目前使用的一些工具链系统，工具链系统的功能结构图如下图。

首先，工具链系统包括一个实时计算平台，主要的功能是统一提交作业和一些资源分配以及监控告警，但是实际上无论是否开发数仓，大概都需要这样的一个工具，这是开发 Flink 的基本工具。

对于我们来讲，跟数仓相关的主要工具有两块:

系统管理模块，这个模块实际上是我们的实时和离线是一起使用的。其中知识库管理模块，主要是用来记录模型中表和字段的一些信息，另外就是一些工单的解决方法也会维护进去。Flink 管理主要是用来管理一些我们公司自己开发的一些 Flink 相关的系统组件。

重点其实还是我们整个用来开发实时数仓 ETL 的一个 开发工具。主要是如下几点：

SQL 及 UDF 管理，管理 SQL 脚本和 UDF，以及对 UDF 进行配置。

任务日志查看和任务监控。

调度管理，主要是管理任务的重导和重传。

数据资产管理，管理实时和离线的元数据，以及任务依赖信息。

其实整个这条工具链，每个工具都有它自己特定的用场场景，下面重点讲解其中两个。

■ 元数据管理

我们在 Flink SQL 的开发过程中，每一个任务都要重新把元数据重新写一遍。因为 kafka 以及很多的缓存组件，如 Tair、Redis 都不支持元数据的管理，所以我们一定要尽早建设元数据管理系统。

■ 血缘管理

血缘其实对于实时数仓来讲比较重要，在上文中也提到过，在实时的作业的运维过程当中，一旦对自己的作业进行了修改，必须保证下游都是能够准确的解析新数据的这样一个情况。如果是依赖于这种人脑去记忆，比如说谁用我的销售表或者口头通知这种方式来讲的话，效率会非常的低，所以一定要建立一套就是血缘的管理机制。要知道到底是谁用了生产的表，然后上游用了谁的，方便大家再进行修改的时候进行周知，保证我们整个实时数仓的稳定。

元数据和血缘管理系统，最简单的实现方式大概分为以下三点：

通过元数据服务生成 Catalog

首先通过元数据系统，把元数据系统里的元数据信息加载到程序中来，然后生成 Flink Catalog 。这样就可以知道当前作业可以消费哪些表，使用哪些表。

解析 DDL 语句创建更新表

当作业进行一系列操作，最终要输出某张表的时候，解析作业里面关于输出部分的 DDL 代码，创建出新的元数据信息写入到元数据系统。

作业信息和运行状态写入元数据

作业本身的元数据信息以及它的运行状态也会同步到元数据系统里面来，让这些信息来帮助我们建立血缘关系。

最终的系统可以通过数据库来存储这些信息，如果你设计的系统没那么复杂，也可以使用文件来进行存储。重点是需要尽快建立一套这样的系统，不然在后续的开发和运维过程当中都会非常的痛苦。