实时数仓架构

背景：

传统意义上的数据仓库主要处理T+1数据（即：今天产生的数据分析结果明天才能看到）。

随着互联网的发展，越来越多的业务指标需要实时查看，以便于更好的进行业务分析，尤其在举行活动的时候，能够更好的把握活动的各项指标趋势，从而更好的调整策略，达到活动的目标。

随着数据时效性在企业运营中的重要性日益凸现，例如，实时推荐、精准营销、广告投放效果、实时物流等。数据的实时处理能力成为企业提升竞争力的一大因素，最初阶段企业主要采用来一个需求，编写一个实时计算任务的方式来处理实时数据，随着需求的增多，计算任务也相应增多，并且不同任务的开发人员不同，导致开发风格差异化，该阶段的实时数据处理缺乏统一的规划，代码风格差异化严重，在维护成本和开发效率上有很大障碍。

为避免上述问题，人们参照数据仓库的概念和模型来重新规划和设计实时数据处理，在此基础上产生了实时数据仓库（实时数仓）。

与离线数仓的对比：

首先，从架构上，实时数仓与离线数仓有比较明显的区别，实时数仓以Kappa架构为主，而离线数仓以传统大数据架构为主。Lambda架构可以认为是两者的中间态。

其次，从建设方法上，实时数仓和离线数仓基本还是沿用传统的数仓主题建模理论，产出事实宽表。另外实时数仓中实时流数据的join有隐藏时间语义，在建设中需注意。

最后，从数据保障看，实时数仓因为要保证实时性，所以对数据量的变化较为敏感。在大促等场景下需要提前做好压测和主备保障工作，这是与离线数据的一个较为明显的区别。

LAMBDA与KAPPA架构的区别：

Lambda 架构

Lambda 架构（Lambda Architecture）是由 Twitter 工程师南森·马茨（Nathan Marz）提出的大数据处理架构。这一架构的提出基于马茨在 BackType 和 Twitter 上的分布式数据处理系统的经验。
Lambda 架构使开发人员能够构建大规模分布式数据处理系统。它具有很好的灵活性和可扩展性，也对硬件故障和人为失误有很好的容错性。

Lambda 架构总共由三层系统组成：批处理层（Batch Layer），速度处理层（Speed Layer），以及用于响应查询的服务层（Serving Layer）。
在 Lambda 架构中，每层都有自己所肩负的任务。
批处理层存储管理主数据集（不可变的数据集）和预先批处理计算好的视图。
批处理层使用可处理大量数据的分布式处理系统预先计算结果。它通过处理所有的已有历史数据来实现数据的准确性。这意味着它是基于完整的数据集来重新计算的，能够修复任何错误，然后更新现有的数据视图。输出通常存储在只读数据库中，更新则完全取代现有的预先计算好的视图。
速度处理层会实时处理新来的大数据。
速度层通过提供最新数据的实时视图来最小化延迟。速度层所生成的数据视图可能不如批处理层最终生成的视图那样准确或完整，但它们几乎在收到数据后立即可用。而当同样的数据在批处理层处理完成后，在速度层的数据就可以被替代掉了。
本质上，速度层弥补了批处理层所导致的数据视图滞后。比如说，批处理层的每个任务都需要 1 个小时才能完成，而在这 1 个小时里，我们是无法获取批处理层中最新任务给出的数据视图的。而速度层因为能够实时处理数据给出结果，就弥补了这 1 个小时的滞后。
所有在批处理层和速度层处理完的结果都输出存储在服务层中，服务层通过返回预先计算的数据视图或从速度层处理构建好数据视图来响应查询。
例如广告投放预测这种推荐系统一般都会用到Lambda架构。一般能做精准广告投放的公司都会拥有海量用户特征、用户历史浏览记录和网页类型分类这些历史数据的。业界比较流行的做法有在批处理层用Alternating Least Squares (ALS)算法，也就是Collaborative Filtering协同过滤算法，可以得出与用户特性一致其他用户感兴趣的广告类型，也可以得出和用户感兴趣类型的广告相似的广告，而用k-means也可以对客户感兴趣的广告类型进行分类。
这里的结果是批处理层的结果。在速度层中根据用户的实时浏览网页类型在之前分好类的广告中寻找一些top K的广告出来。最终服务层可以结合速度层的top K广告和批处理层中分类好的点击率高的相似广告，做出选择投放给用户。
Lambda 架构的不足
虽然 Lambda 架构使用起来十分灵活，并且可以适用于很多的应用场景，但在实际应用的时候，Lambda 架构也存在着一些不足，主要表现在它的维护很复杂。
使用 Lambda 架构时，架构师需要维护两个复杂的分布式系统，并且保证他们逻辑上产生相同的结果输出到服务层中。
我们都知道，在分布式框架中进行编程其实是十分复杂的，尤其是我们还会针对不同的框架进行专门的优化。所以几乎每一个架构师都认同，Lambda 架构在实战中维护起来具有一定的复杂性。
那要怎么解决这个问题呢？我们先来思考一下，造成这个架构维护起来如此复杂的根本原因是什么呢？
维护 Lambda 架构的复杂性在于我们要同时维护两套系统架构：批处理层和速度层。我们已经说过了，在架构中加入批处理层是因为从批处理层得到的结果具有高准确性，而加入速度层是因为它在处理大规模数据时具有低延时性。
那我们能不能改进其中某一层的架构，让它具有另外一层架构的特性呢？
例如，改进批处理层的系统让它具有更低的延时性，又或者是改进速度层的系统，让它产生的数据视图更具准确性和更加接近历史数据呢？
另外一种在大规模数据处理中常用的架构——Kappa 架构（Kappa Architecture），便是在这样的思考下诞生的。
Kappa 架构
Kappa 架构是由 LinkedIn 的前首席工程师杰伊·克雷普斯（Jay Kreps）提出的一种架构思想。克雷普斯是几个著名开源项目（包括 Apache Kafka 和 Apache Samza 这样的流处理系统）的作者之一，也是现在 Confluent 大数据公司的 CEO。
克雷普斯提出了一个改进 Lambda 架构的观点：
我们能不能改进 Lambda 架构中速度层的系统性能，使得它也可以处理好数据的完整性和准确性问题呢？我们能不能改进 Lambda 架构中的速度层，使它既能够进行实时数据处理，同时也有能力在业务逻辑更新的情况下重新处理以前处理过的历史数据呢？
他根据自身多年的架构经验发现，我们是可以做到这样的改进的。
像 Apache Kafka 这样的流处理平台是具有永久保存数据日志的功能的，通过平台的这一特性，我们可以重新处理部署于速度层架构中的历史数据。
下面就以 Apache Kafka 为例来讲述整个全新架构的过程。

第一步，部署 Apache Kafka，并设置数据日志的保留期（Retention Period）。这里的保留期指的是你希望能够重新处理的历史数据的时间区间。
例如，如果你希望重新处理最多一年的历史数据，那就可以把 Apache Kafka 中的保留期设置为 365 天。如果你希望能够处理所有的历史数据，那就可以把 Apache Kafka 中的保留期设置为“永久（Forever）”。
第二步，如果我们需要改进现有的逻辑算法，那就表示我们需要对历史数据进行重新处理。
我们需要做的就是重新启动一个 Apache Kafka 作业实例（Instance）。这个作业实例将从头开始，重新计算保留好的历史数据，并将结果输出到一个新的数据视图中。我们知道 Apache Kafka 的底层是使用 Log Offset 来判断现在已经处理到哪个数据块了，所以只需要将 Log Offset 设置为 0，新的作业实例就会从头开始处理历史数据。
第三步，当这个新的数据视图处理过的数据进度赶上了旧的数据视图时，我们的应用便可以切换到从新的数据视图中读取。
第四步，停止旧版本的作业实例，并删除旧的数据视图。
与 Lambda 架构不同的是，Kappa 架构去掉了批处理层这一体系结构，而只保留了速度层。你只需要在业务逻辑改变又或者是代码更改的时候进行数据的重新处理。
在讲述完 Kappa 架构之后，我想强调一下，Kappa 架构也是有着它自身的不足的。
因为 Kappa 架构只保留了速度层而缺少批处理层，在速度层上处理大规模数据可能会有数据更新出错的情况发生，这就需要我们花费更多的时间在处理这些错误异常上面。
还有一点，Kappa 架构的批处理和流处理都放在了速度层上，这导致了这种架构是使用同一套代码来处理算法逻辑的。所以 Kappa 架构并不适用于批处理和流处理代码逻辑不一致的场景。

总结

Lambda 架构和 Kappa 架构这两种大规模数据处理架构，它们都各自有着自身的优缺点。我们需要按照实际情况来权衡利弊，看看在业务中到底需要使用到哪种架构。在文中中主要讲述的是LAMBDA架构的相关知识。

实现架构与组件选型：

Lambda架构应具备的特征

• 健壮和容错
  

分布式数据库要保证数据的一致性、避免重复数据、并发等。同时，还要保证数据总是正确的，这会给系统设计带来较大挑战。我们应该尽量避免这种复杂。同时要具备人为容错能力。也就是说需要避免因为人为的失误，而导致数据丢失、数据错误。需要为系统提供重跑数据的能力，能够在人为导致数据错误情况，重新获得正确的数据。同时，保证数据的不可变性，外部程序无法修改数据。

• 低延迟读和更新
  

系统能够不影响系统健壮性的同时实现低延迟读取和更新。大家看这些年的技术发展，相信有很多人都想将低延迟读取和更新融合到HDFS中。而其实十几年前就有这样的实践，例如：HBase。虽然，HBase是属于hadoop生态，但就数据模型、存储引擎来说，HBase做了非常大的调整，只不过在底层，数据仍然是保存在HDFS中，其他的基本上跟HDFS差距非常大。所以，HBase也划入到了NoSQL的阵营。
要让数据系统能够支持各种各样的应用场景，我们是否能够依赖一个组件就能完成所有呢？如果真的可以，那Lambda架构让组件相互协作的方式将会被打破。但目前，其实我们并没有看到在大数据领域中有这样的一款万精油组件。其实，在OLTP系统中，也没有这样的组件。就像我们会用消息队列来进行流量削峰，我们会使用高速缓存在减轻数据库的负载，我们会使用全文检索引擎来加速模糊、关联查询等等。且看IT这么多年的发展，我根本不相信会有这样的组件出现。
说到这里，需要来聊一个比较另类的存储系统——Kudu。虽然Kudu也是一个LSM结构的存储引擎，但它的数据模型看起来像极了关系型模型。但如果用过Kudu就会知道，Kudu本身能够支持的操作非常少，只是它有表的概念、有主键、列的概念。但如果要进行一些类似于关系型的复杂操作，我们得借助于Impala或者是Spark这样的引擎。
Kudu官方也介绍，它是一款介于HDFS、NoSQL之间的存储系统。Kudu也确实实现了能够以低延迟写入、更新以及主键查询。而在OLAP上。但如果用Kudu存储海量的数据、以及进行复杂一点的分析，我们会发现，在OLAP上Kudu还是弱了很多，因为它的结构非常复杂，要兼顾事务、并发、低延迟读写、数据分析，所以，它确实也是一个中间型的产品。维护的成本比较高，资源消耗也会比较大。大规模的数据探索，Kudu有非常大的局限性。而且写放大问题很难解决。这些问题，当在海量数据面前，全部都会暴露出来。

• 可扩展性
  

在数据负载增加时，可以通过向系统增加资源来保证性能。可扩展性几乎是数据系统中的必备选项。当然，我们不能排除所有的公司都是这样的。例如：一个公司业务急剧下滑，数据突然急剧减少。这种情况时存在的，但我坚信，如果一个公司出现这样的情况，公司会想办法来降低数据系统的开发、维护成本。而将更多地资源投入到业务中，技术人员的价值也会逐渐衰落。我相信，没有哪个架构师愿意看到这样的局面，更没有哪个架构师会基于这样的场景来去设计系统。
Lambda架构能够保障每个层都可以进行水平扩展，也就是添加更多的机器来实现扩展。水平扩展，是系统扩展的常见方式。而我们基于大数据组件开发的应用，基本上也都是能够水平扩展的。没有人会将一个用于调试的单机系统部署到生产环境中。
当添加新功能或者对系统功能进行调整时，可以对系统进行扩展，并且能够以最小的成本来开发。例如：能够快速地进行数据迁移、格式转换等。统一是目前数据系统建设的重要原则。为减小开发成本，我们在设计之初尽可能地减少架构的复杂度、减少引入不同的存储体系、和计算体系。每当我们添加一个组件、每当我们添加一个模块，都会提升开发成本，开发成本提升了，维护成本也上来了。

• 即席查询
  

要为挖掘数据提供即席查询的能力。即席查询对开发人员、分析人员或者是维护人员都很重要。不要指望在企业中能够让ETL开发工程师提供所有的数据、和计算结果。一个尊重数据的公司，一个谋求发展的公司，不会指望ETL开发工程师能够带来太多的业务创新，以及去发现、解决业务问题。我们需要给这些用户去探索数据的空间。那么AdHoc是必不可少的。
AdHoc的引擎有Presto、Impala等。但AdHoc也不一定就是MPP引擎的独家授权。为了能够更好的管控资源，稳定的性能。Spark、Flink也可以提供有一定延迟的AdHoc。这没有什么问题。毕竟，如果定位用户为分析人员、开发人员、或者维护人员，大多数的场景也都不是毫秒级、秒级的。

• 最小的维护成本
  

选择一个功能实现比较简单，复杂度不高的组件。系统越复杂，出现问题的概率也就越大。维护成本我们需要考虑下公司的运维人员，当我们研发完毕后，要将数据系统的维护移交给运维人员。让他们来完成一些操作。不要指望运维人员的技术能力有多强，我们设计的数据系统就应该能够降低运维门槛。维护成本越低，系统出现故障的概率、人为的错误操作带来的风险，也都会下降。
而Lambda架构可以将复杂性从核心组件推到系统的各个部分。

• 可调试性
  

当出现问题时，能够提供更多的调试信息。能够跟踪系统中的每个值，确切地知道是什么导致问题。可调试性也是需要Lambda架构中保证的，而且能够使用数据重跑的方式重新计算数据。
可调试性在系统的设计中非常关键，但系统出现一些问题时，我们需要重新追数、重新跑数。或者我们想要查找到是在什么时间、什么数据、什么操作下出现的问题。并且能够追溯到每个细节。这就要求在我们设计应用程序（不限于ETL、实时应用等）时，我们需要考虑可调试。我们需要给系统做较多的配套工具，通过工具能够快速发现问题、解决问题。

如何建设：

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 里的这些维度可以尽早的被淘汰掉。因为很多天以前的维度，实际上也不会再关联了，这样就保证维度数据不会无限制的增长，导致存储爆炸。
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 到一个实时数据当时的准确的维度。