1、实时数仓建设目的算法
一、解决传统数仓的问题
实时数仓是一个很容易让人产生混淆的概念。实时数仓自己彷佛和把 PPT 黑色的背景变得更白同样，从传统的经验来说，咱们认为数仓有一个很重要的功能，即可以记录历史。一般，数仓都是但愿从业务上线的第一天开始有数据，而后一直记录到如今。数据库
但实时处理技术，又是强调当前处理状态的一门技术，因此咱们认为这两个相对对立的方案重叠在一块儿的时候，它注定不是用来解决一个比较普遍问题的一种方案。因而，咱们把实时数仓建设的目的定位为解决因为传统数据仓库数据时效性低解决不了的问题。编程
因为这个特色，咱们给定了两个原则：缓存
-
传统数仓能解决的问题，实时数仓就不解决了。好比上个月的一些历史的统计，这些数据是不会用实时数仓来建设的。架构
-
问题自己就不太适合用数仓来解决，也不用实时数仓解决。好比业务性很强的需求，或者是对时效性要求特别高的需求。这些需求咱们也不建议经过实时数仓这种方式来进行解决。app
固然为了让咱们整个系统看起来像是一个数仓，咱们仍是给本身提了一些要求的。这个要求其实跟咱们创建离线数仓的要求是同样的，首先实时的数仓是须要面向主题的，而后具备集成性，而且保证相对稳定。框架
离线数仓和实时数仓的区别在于离线数据仓库是一个保存历史累积的数据，而咱们在建设实时数仓的时候，咱们只保留上一次批处理到当前的数据。这个说法很是的拗口，可是实际上操做起来仍是蛮轻松的。运维
一般来说解决方案是保留大概三天的数据，由于保留三天的数据的话，能够稳定地保证两天完整的数据，这样就能保证，在批处理流程尚未处理完昨天的数据的这段间隙，依然可以提供一个完整的数据服务。工具
二、实时数仓的应用场景
1）实时 OLAP 分析oop
OLAP 分析自己就很是适合用数仓去解决的一类问题，咱们经过实时数仓的扩展，把数仓的时效性能力进行提高。甚至可能在分析层面上都不用再作太多改造，就可使原有的 OLAP 分析工具具备分析实时数据的能力。
2）实时数据看板
这种场景比较容易接受，好比天猫双11的实时大屏滚动展现核心数据的变化。实际上对于美团来说，不光有促销上的业务，还有一些主要的门店业务。对于门店的老板而言，他们可能在平常的每一天中也会很关心本身当天各个业务线上的销售额。
3）实时特征
实时特征指经过汇总指标的运算来对商户或者用户标记上一些特征。好比屡次购买商品的用户后台会断定为优质用户。另外，商户销售额稿，后台会认为该商户商的热度更高。而后，在作实时精准运营动做时可能会优先考虑相似的门店或者商户。
4）实时业务监控
美团点评也会对一些核心业务指标进行监控，好比说当线上出现一些问题的时候，可能会致使某些业务指标降低，咱们能够经过监控尽早发现这些问题，进而来减小损失。
2、如何建设实时数仓
一、实时数仓概念映射
咱们经过离线数仓开发和实时数仓开发的对应关系表，帮助你们快速清晰的理解实时数仓的一些概念。
1）编程方式
离线开发最多见的方案就是采用 Hive SQL 进行开发，而后加上一些扩展的 udf 。映射到实时数仓里来，咱们会使用 Flink SQL ，一样也是配合 udf 来进行开发。
2）做业执行层面
离线处理的执行层面通常是 MapReduce 或者 Spark Job ，对应到实时数仓就是一个持续不断运行的 Flink Streaming 的程序。
3）数仓对象层面
离线数仓实际上就是在使用 Hive 表。对于实时数仓来说，咱们对表的抽象是使用 Stream Table 来进行抽象。
4）物理存储
离线数仓，咱们多数状况下会使用 HDFS 进行存储。实时数仓，咱们更多的时候会采用像 Kafka 这样的消息队列来进行数据的存储。
二、实时数仓的总体架构
在此以前咱们作过一次分享，是关于为何选择 Flink 来作实时数仓，其中重点介绍了技术组件选型的缘由和思路，具体内容参考《美团点评基于 Flink 的实时数仓建设实践》。本文分享的主要内容是围绕数据自己来进行的，下面是咱们目前的实时数仓的数据架构图：
从数据架构图来看，实时数仓的数据架构会跟离线数仓有不少相似的地方。好比分层结构；好比说 ODS 层，明细层、汇总层，乃至应用层，它们命名的模式可能都是同样的。尽管如此，实时数仓和离线数仓仍是有不少的区别的。
跟离线数仓主要不同的地方，就是实时数仓的层次更少一些。
以咱们目前建设离线数仓的经验来看，数仓的第二层远远不止这么简单，通常都会有一些轻度汇总层这样的概念，其实第二层会包含不少层。另一个就是应用层，以往建设数仓的时候，应用层实际上是在仓库内部的。在应用层建设好后，会建同步任务，把数据同步到应用系统的数据库里。
在实时数仓里面，所谓 APP 层的应用表，实际上就已经在应用系统的数据库里了。上图，虽然画了 APP 层，但它其实并不算是数仓里的表，这些数据本质上已经存过去了。
为何主题层次要少一些？是由于在实时处理数据的时候，每建一个层次，数据必然会产生必定的延迟。
为何汇总层也会尽可能少建？是由于在汇总统计的时候，每每为了容忍一部分数据的延迟，可能会人为的制造一些延迟来保证数据的准确。
举例，统计事件中的数据时，可能会等到 10:00:05 或者 10:00:10再统计，确保 10:00 前的数据已经所有接受到位了，再进行统计。因此，汇总层的层次太多的话，就会更大的加剧人为形成的数据延迟。
建议尽可能减小层次，特别是汇总层必定要减小，最好不要超过两层。明细层可能多一点层次还好，会有这种系统明细的设计概念。
第二个比较大的不一样点就是在于数据源的存储。
在建设离线数仓的时候，可能整个数仓都所有是创建在 Hive 表上，都是跑在 Hadoop 上。可是，在建设实时数仓的时候，同一份表，咱们甚至可能会使用不一样的方式进行存储。
好比常见的状况下，可能绝大多数的明细数据或者汇总数据都会存在 Kafka 里面，可是像维度数据，可能会存在像 Tair 或者 HBase 这样的 kv 存储的系统中，实际上可能汇总数据也会存进去，具体缘由后面详细分析。除了总体结构，咱们也分享一下每一层建设的要点。
1）ODS 层的建设
-
数据来源尽量统一
-
利用分区保证数据局部有序
首先第一个建设要点就是 ODS 层，其实 ODS 层建设可能跟仓库不必定有必然的关系，只要使用 Flink 开发程序，就必然都要有实时的数据源。目前主要的实时数据源是消息队列，如 Kafka。而咱们目前接触到的数据源，主要仍是以 binlog、流量日志和系统日志为主。
这里面我主要想讲两点：
首先第一个建设要点就是 ODS层，其实ODS层建设可能跟这个仓库不必定有必然的关系，只要你使用这个flink开发程序，你必然都要有这种实时的数据源。目前的主要的实时数据源就是消息队列，如kafka。咱们目前接触到的数据源，主要仍是以binlog、流量日志和系统日志为主。
这里面我主要想讲两点，一个这么多数据源我怎么选？咱们认为以数仓的经验来看：
首先就是数据源的来源尽量要统一。这个统一有两层含义：
第一个统一就是实时的数据源自己要跟本身统一，好比你选择从某个系统接入某一种数据，要么都从binlog来接，要么都从系统日志来接，最好不要混着接。在不知道数据生产的流程的状况下，一部分经过binlog接入一部分经过系统日志接入，容易出现数据乱序的问题。
第二个统一是指实时和离线的统一，这个统一可能更重要一点。虽然咱们是建设实时数仓，可是本质上仍是数仓，做为一个团队来说，仓库里的指标的计算逻辑和数据来源应该彻底一致，不能让使用数据的人产生误解。若是一个数据两个团队都能为你提供，咱们建议选择跟离线同窗一致的数据来源。包括咱们公司自己也在作一些保证离线和实时采用的数据源一致的工做。
第二个要点就是数据乱序的问题，咱们在采集数据的时候会有一个比较大的问题，可能同一条数据，因为分区的存在，这条数据先发生的状态后消费到，后发生的状态先消费到。咱们在解决这一问题的时候采用的是美团内部的一个数据组件。
其实，保证数据有序的主要思路就是利用 kafka 的分区来保证数据在分区内的局部有序。至于具体如何操做，能够参考《美团点评基于 Flink 的实时数仓建设实践》。这是咱们美团数据同步部门作的一套方案，能够提供很是丰富的策略来保证同一条数据是按照生产顺序进行保序消费的，实如今源头解决数据乱序的问题。
2）DW 层的建设
解决原始数据中数据存在噪声、不完整和数据形式不统一的状况。造成规范，统一的数据源。若是可能的话尽量和离线保持一致。
明细层的建设思路其实跟离线数仓的基本一致，主要在于如何解决 ODS 层的数据可能存在的数据噪声、不完整和形式不统一的问题，让它在仓库内是一套知足规范的统一的数据源。咱们的建议是若是有可能的话，最好入什么仓怎么入仓，这个过程和离线保持一致。
尤为是一些数据来源比较统一，可是开发的逻辑常常变化的系统，这种状况下，咱们可能采用的实际上是一套基于配置的入仓规则。可能离线的同窗有一套入仓的系统，他们配置好规则就知道哪些数据表上数据要进入实时数仓，以及要录入哪些字段，而后实时和离线是采用同一套配置进行入仓，这样就能够保证咱们的离线数仓和实时数仓在 DW 层长期保持一个一致的状态。
实际上建设 DW 层其实主要的工做主要是如下4部分：
惟一标红的就是模型的规范化，其实模型的规范化，是一个老生常谈的问题，可能每一个团队在建设数仓以前，都会先把本身的规范化写出来。但实际的结果是咱们会看到其实并非每个团队最终都能把规范落地。
在实时的数仓建设当中，咱们要特别强调模型的规范化，是由于实施数仓有一个特色，就是自己实时做业是一个7×24 小时调度的状态，因此当修改一个字段的时候，可能要付出的运维代价会很高。在离线数仓中，可能改了某一个表，只要一天以内把下游的做业也改了，就不会出什么问题。可是实时数仓就不同了，只要改了上游的表结构，下游做业必须是可以正确解析上游数据的状况下才能够。
另外使用像 kafka 这样的系统，它自己并非结构化的存储，没有元数据的概念，也不可能像改表同样，直接把以前不规范的表名、表类型改规范。要在过后进行规范代价会很大。因此建议必定要在建设之初就尽快把这些模型的规范化落地，避免后续要投入很是大的代价进行治理。
3）重复数据处理
除了数据自己咱们会在每条数据上额外补充一些信息，应对实时数据生产环节的一些常见问题：
4）惟一键和主键
咱们会给每一条数据都补充一个惟一键和一个主键，这两个是一对的，惟一键就是标识是惟一一条数据的，主键是标记为一行数据。一行数据可能变化不少次，可是主键是同样的，每一次变化都是其一次惟一的变化，因此会有一个惟一键。惟一键主要解决的是数据重复问题，从分层来说，数据是从咱们仓库之外进行生产的，因此很难保证咱们仓库之外的数据是不会重复的。
可能有些人交付数据给也会告知数据可能会有重复。生成惟一键的意思是指咱们须要保证 DW 层的数据可以有一个标识，来解决可能因为上游产生的重复数据致使的计算重复问题。生成主键，其实最主要在于主键在 kafka 进行分区操做，跟以前接 ODS 保证分区有序的原理是同样的，经过主键，在 kafka 里进行分区以后，消费数据的时候就能够保证单条数据的消费是有序的。
5）版本和批次
版本和批次这两个其实又是一组。固然这个内容名字能够随便起，最重要的是它的逻辑。
首先，版本。版本的概念就是对应的表结构，也就是 schema 一个版本的数据。因为在处理实时数据的时候，下游的脚本依赖表上一次的 schema 进行开发的。当数据表结构发生变化的时候，就可能出现两种状况：第一种状况，可能新加或者删减的字段并无用到，其实彻底不用感知，不用作任何操做就能够了。另一种状况，须要用到变更的字段。此时会产生一个问题，在 Kafka 的表中，就至关于有两种不一样的表结构的数据。这时候其实须要一个标记版本的内容来告诉咱们，消费的这条数据到底应该用什么样的表结构来进行处理，因此要加一个像版本这样的概念。
第二，批次。批次其实是一个更不常见的场景，有些时候可能会发生数据重导，它跟重启不太同样，重启做业可能就是改一改，而后接着上一次消费的位置启动。而重导的话，数据消费的位置会发生变化。
好比，今天的数据算错了，领导很着急让我改，而后我须要把今天的数据重算，可能把数据程序修改好以后，还要设定程序，好比从今天的凌晨开始从新跑。这个时候因为整个数据程序是一个 7x24 小时的在线状态，其实原先的数据程序不能停，等重导的程序追上新的数据以后，才能把原来的程序停掉，最后使用重导的数据来更新结果层的数据。
在这种状况下，必然会短暂的存在两套数据。这两套数据想要进行区分的时候，就要经过批次来区分。其实就是全部的做业只消费指定批次的数据，当重导做业产生的时候，只有消费重导批次的做业才会消费这些重导的数据，而后数据追上以后，只要把原来批次的做业都停掉就能够了，这样就能够解决一个数据重导的问题。
6）维度数据建设
其次就是维度数据，咱们的明细层里面包括了维度数据。关于维度的数据的处理，其实是先把维度数据分红了两大类采用不一样的方案来进行处理。
① 变化频率低的维度
第一类数据就是一些变化频率比较低的数据，这些数据其实多是一些基本上是不会变的数据。好比说，一些地理的维度信息、节假日信息和一些固定代码的转换。
这些数据实际上咱们采用的方法就是直接能够经过离线仓库里面会有对应的维表，而后经过一个同步做业把它加载到缓存中来进行访问。还有一些维度数据建立得会很快，可能会不断有新的数据建立出来，可是一旦建立出来，其实也就再也不会变了。
好比说，美团上开了一家新的门店，门店所在的城市名字等这些固定的属性，其实可能很长时间都不会变，取最新的那一条数据就能够了。这种状况下，咱们会经过公司内部的一些公共服务，直接去访问当前最新的数据。最终，咱们会包一个维度服务的这样一个概念来对用户进行屏蔽，具体是从哪里查询相关细节，经过维度服务便可关联具体的维度信息。
② 变化频率高的维度
第二类是一些变化频率较高的数据。好比常见的病人心脑科的状态变更，或者某一个商品的价格等。这些东西每每是会随着时间变化比较频繁，比较快。而对于这类数据，咱们的处理方案就稍微复杂一点。首先对于像价格这样变化比较频繁的这种维度数据，会监听它的变化。好比说，把价格想象成维度，咱们会监听维度价格变化的消息，而后构建一张价格变换的拉链表。
一旦创建了维度拉链表，当一条数据来的时候，就能够知道，在这个数据某一时刻对应的准确的维度是多少，避免了因为维度快速的变化致使关联错维度的问题。
另外一类如新老客这维度，于咱们而言实际上是一种衍生维度，由于它自己并非维度的计算方式，是用该用户是否下过单来计算出来的，因此它实际上是用订单数据来算出来的一个维度。
因此相似订单数的维度，咱们会在 DW 层创建一些衍生维度的计算模型，而后这些计算模型输出的其实也是拉链表，记录下一个用户天天这种新老客的变化程度，或者多是一个优质用户的变化的过程。因为创建拉链表自己也要关联维度，因此能够经过以前分组 key 的方式来保障不乱序，这样仍是将其当作一个不变的维度来进行关联。
经过这种方式来创建拉链表相对麻烦，因此实际上建议利用一些外部组件的功能。实际操做的时候，咱们使用的是 Hbase。HBase 自己支持数据多版本的，并且它能记录数据更新的时间戳，取数据的时候，甚至能够用这个时间戳来作索引。
因此实际上只要把数据存到 HBase 里，再配合上 mini-versions ，就能够保证数据不会超时死掉。上面也提到过，整个实时数仓有一个大原则，不处理离线数仓能处理的过程。至关于处理的过程，只须要处理三天之内的数据，因此还能够经过配置 TTL 来保证 HBase 里的这些维度能够尽早的被淘汰掉。由于数日之前的维度，实际上也不会再关联了，这样就保证维度数据不会无限制的增加，致使存储爆炸。
7）维度数据使用
处理维度数据以后，这个维度数据怎么用？
第一种方案，也是最简单的方案，就是使用 UDTF 关联。其实就是写一个 UDTF 去查询上面提到的维度服务，具体来说就是用 LATERAL TABLE 关键词来进行关联，内外关联都是支持的。
另一种方案就是经过解析 SQL ，识别出关联的维表以及维表中的字段，把它本来的查询进行一次转化为原表.flatmap (维表)，最后把整个操做的结果转换成一张新的表来完成关联操做。
可是这个操做要求使用者有不少周边的系统来进行配合，首先须要能解析 SQL ，同时还能识别文本，记住全部维表的信息，最后还要能够执行 SQL 转化，因此这套方案适合一些已经有成熟的基于 Flink SQL 的 SQL开发框架的系统来使用。若是只是单纯的写封装的代码，建议仍是使用 UDTF 的方式来进行关联会很是的简单，并且效果也是同样的。
8）汇总层的建设
在建设实时数仓的汇总层的时候，跟离线的方案其实会有不少同样的地方。
第一点是对于一些共性指标的加工，好比说 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 到一个实时数据当时的准确的维度。
3、仓库质量保证
通过上面的环节，若是你已经创建好了一个仓库，你会发现想保证仓库的正常的运行或者是保证它高质量的运行，实际上是一个很是麻烦的过程，它要比一线的操做复杂得多，因此咱们在建设完仓库以后，须要建设不少的周边系统来提升咱们的生产效率。
下面介绍一下咱们目前使用的一些工具链系统，工具链系统的功能结构图以下图：
首先，工具链系统包括一个实时计算平台，主要的功能是统一提交做业和一些资源分配以及监控告警，可是实际上不管是否开发数仓，大概都须要这样的一个工具，这是开发 Flink 的基本工具。
对于咱们来说，跟数仓相关的主要工具备两块:
1）系统管理模块，这个模块其实是咱们的实时和离线是一块儿使用的。其中知识库管理模块，主要是用来记录模型中表和字段的一些信息，另外就是一些工单的解决方法也会维护进去。Flink 管理主要是用来管理一些咱们公司本身开发的一些 Flink 相关的系统组件。
2）重点其实仍是咱们整个用来开发实时数仓 ETL 的一个开发工具。主要是以下几点：
-
SQL 及 UDF 管理，管理 SQL 脚本和 UDF，以及对 UDF 进行配置。
-
任务日志查看和任务监控。
-
调度管理，主要是管理任务的重导和重传。
-
数据资产管理，管理实时和离线的元数据，以及任务依赖信息。
其实整个这条工具链，每一个工具都有它本身特定的用场场景，下面重点讲解其中两个。
一、元数据与血缘管理
1）元数据管理
咱们在 Flink SQL 的开发过程当中，每个任务都要从新把元数据从新写一遍。由于 kafka 以及不少的缓存组件，如 Tair、Redis 都不支持元数据的管理，因此咱们必定要尽早建设元数据管理系统。
2）血缘管理
血缘其实对于实时数仓来说比较重要，在上文中也提到过，在实时的做业的运维过程中，一旦对本身的做业进行了修改，必须保证下游都是可以准确的解析新数据的这样一个状况。若是是依赖于这种人脑去记忆，好比说谁用个人销售表或者口头通知这种方式来说的话，效率会很是的低，因此必定要创建一套就是血缘的管理机制。要知道究竟是谁用了生产的表，而后上游用了谁的，方便你们再进行修改的时候进行周知，保证咱们整个实时数仓的稳定。
元数据和血缘管理系统，最简单的实现方式大概分为如下三点：
① 经过元数据服务生成 Catalog
首先经过元数据系统，把元数据系统里的元数据信息加载到程序中来，而后生成 Flink Catalog 。这样就能够知道当前做业能够消费哪些表，使用哪些表。
② 解析 DDL 语句建立更新表
看成业进行一系列操做，最终要输出某张表的时候，解析做业里面关于输出部分的 DDL 代码，建立出新的元数据信息写入到元数据系统。
③ 做业信息和运行状态写入元数据
做业自己的元数据信息以及它的运行状态也会同步到元数据系统里面来，让这些信息来帮助咱们创建血缘关系。
最终的系统能够经过数据库来存储这些信息，若是你设计的系统没那么复杂，也可使用文件来进行存储。重点是须要尽快创建一套这样的系统，否则在后续的开发和运维过程中都会很是的痛苦。
二、数据质量验证
将实时数据写入 Hive，使用离线数据持续验证明时数据的准确性。
当建设完一个数仓以后，尤为是第一次创建以后，必定会很是怀疑本身数据到底准不许。在此以前的验证方式就是经过写程序去仓库里去查，而后来看数据对不对。在后续的建设过程当中咱们发现天天这样人为去对比太累了。
咱们就采起了一个方案，把中间层的表写到 Hive 里面去，而后利用离线数据丰富的质量验证工具去对比离线和实时同一模型的数据差别，最后根据设定的阈值进行监控报警。这个方案虽然并不能及时的发现实时数据的问题，可是能够帮助你在上线前了解实时模型的准确程度。而后进行任务的改造，不断提升数据的准确率。另外这个方案还能够检验离线数据的准确性。
以上是美团点评基于 Flink 构建的实时数仓应用经验的分享，但愿对你们有所帮助！
做者丨黄伟伦@美团点评 来源丨Flink 中文社区（ID：gh_5efd76d10a8d) dbaplus社群欢迎广大技术人员投稿，投稿邮箱： editor@dbaplus.cn