应用案例 | 美团点评基于 Flink 的实时数仓建设实践

本文是美团技术团队分享的美团点评基于 Flink 的实时数仓建设实践，Apache Flink 社区公众号（ Ververica ）受权转载，文章主要从常见实时数据组件的性能特色和适用场景以及美团经过 Flink 构建实时数据仓库的过程分享其经验，供社区参考。

引言

近些年，企业对数据服务实时化服务需求日益增多。本文整理了常见实时数据组件的性能特色和适用场景，介绍了美团如何经过 Flink 引擎构建实时数据仓库，从而提供高效、稳健的实时数据服务。此前咱们美团技术博客发布过一篇文章《流计算框架 Flink 与 Storm 的性能对比》，对 Flink 和 Storm 两个引擎的计算性能进行了比较。本文主要阐述使用 Flink 在实际数据生产上的经验。

实时平台初期架构

在实时数据系统建设初期，业务需求也相对较少，尚未造成完整的数据体系。咱们采用的是“一路到底”的开发模式：经过在实时计算平台上部署 Storm 做业处理实时数据队列来提取数据指标，直接推送到实时应用服务中。

可是，随着产品和业务人员对实时数据需求的不断增多，新的挑战也随之发生。

数据指标愈来愈多，“烟囱式”的开发致使代码耦合问题严重。 需求愈来愈多，有的须要明细数据，有的须要 OLAP 分析。单一的开发模式难以应付多种需求。 缺乏完善的监控系统，没法在对业务产生影响以前发现并修复问题。

实时数据仓库的构建

为解决以上问题，咱们根据生产离线数据的经验，选择使用分层设计方案来建设实时数据仓库，其分层架构以下图所示：

该方案由如下四层构成：

1. ODS 层：Binlog 和流量日志以及各业务实时队列。

2. 数据明细层：业务领域整合提取事实数据，离线全量和实时变化数据构建实时维度数据。

3. 数据汇总层：使用宽表模型对明细数据补充维度数据，对共性指标进行汇总。

4. App 层：为了具体需求而构建的应用层，经过 RPC 框架对外提供服务。

经过多层设计咱们能够将处理数据的流程沉淀在各层完成。好比在数据明细层统一完成数据的过滤、清洗、规范、脱敏流程；在数据汇总层加工共性的多维指标汇总数据，提升了代码的复用率和总体生产效率。同时各层级处理的任务类型类似，能够采用统一的技术方案优化性能，使数仓技术架构更简洁。

技术选型

1. 存储引擎的调研

实时数仓在设计中不一样于离线数仓在各层级使用同种储存方案，好比都存储在 Hive 、DB 中的策略。首先对中间过程的表，采用将结构化的数据经过消息队列存储和高速 KV 存储混合的方案。实时计算引擎能够经过监听消息消费消息队列内的数据，进行实时计算；而在高速 KV 存储上的数据则能够用于快速关联计算，好比维度数据。

其次在应用层上，针对数据使用特色配置存储方案直接写入，避免了离线数仓应用层同步数据流程带来的处理延迟。为了解决不一样类型的实时数据需求，合理的设计各层级存储方案，咱们调研了美团内部使用比较普遍的几种存储方案。

根据不一样业务场景，实时数仓各个模型层次使用的存储方案大体以下：

• 数据明细层：对于维度数据部分场景下关联的频率可达 10万多TPS，咱们选择 Cellar（美团内部基于Tair开发的KV存储） 做为存储，封装维度服务为实时数仓提供维度数据。
• 数据汇总层：对于通用的汇总指标，须要进行历史数据关联的数据，采用和维度数据同样的方案经过 Cellar 做为存储，用服务的方式进行关联操做。
• 数据应用层：应用层设计相对复杂，在对比了几种不一样存储方案后。咱们制定了以数据读写频率 1000 QPS 为分界的判断依据。对于读写平均频率高于 1000 QPS 但查询不太复杂的实时应用，好比商户实时的经营数据。采用 Cellar 为存储，提供实时数据服务。对于一些查询复杂的和须要明细列表的应用，使用 Elasticsearch 做为存储则更为合适；而一些查询频率低，好比一些内部运营的数据， Druid 经过实时处理消息构建索引，并经过预聚合能够快速的提供实时数据 OLAP 分析功能。对于一些历史版本的数据产品进行实时化改造时，也可使用 MySQL 存储便于产品迭代。

2. 计算引擎的调研

在实时平台建设初期咱们使用 Storm 引擎来进行实时数据处理。Storm 引擎虽然在灵活性和性能上都表现不错，可是因为 API 过于底层，在数据开发过程当中须要对一些经常使用的数据操做进行功能实现。好比表关联、聚合等，产生了不少额外的开发工做，不只引入了不少外部依赖好比缓存，并且实际使用时性能也不是很理想。同时， Storm 内的数据对象 Tuple 支持的功能也很简单，一般须要将其转换为 Java 对象来处理。

对于这种基于代码定义的数据模型，一般咱们只能经过文档来进行维护，不只须要额外的维护工做，同时在增改字段时也很麻烦。综合来看，使用 Storm 引擎构建实时数仓难度较大。咱们须要一个新的实时处理方案，要可以实现：

1. 提供高级 API，支持常见的数据操做好比关联聚合，最好是能支持 SQL。
2. 具备状态管理和自动支持久化方案，减小对存储的依赖。
3. 便于接入元数据服务，避免经过代码管理数据结构。
4. 处理性能至少要和 Storm 一致。

咱们对主要的实时计算引擎进行了技术调研。总结了各种引擎特性以下表所示：

从调研结果来看，Flink 和 Spark Streaming 的 API 、容错机制与状态持久化机制均可以解决一部分，咱们目前使用 Storm 中遇到的问题；但 Flink 在数据延迟上和 Storm 更接近，对现有应用影响最小，并且在公司内部的测试中 Flink 的吞吐性能对比 Storm 有十倍左右提高。综合考量，咱们选定 Flink 引擎做为实时数仓的开发引擎。

更加引发咱们注意的是，Flink 的 Table 抽象和 SQL 支持，虽然使用 Strom 引擎也能够处理结构化数据，但毕竟依旧是基于消息的处理 API ，在代码层层面上不能彻底享受操做结构化数据的便利。而 Flink 不只支持了大量经常使用的 SQL 语句，基本覆盖了咱们的开发场景，而且 Flink 的 Table 能够经过 TableSchema 进行管理，支持丰富的数据类型和数据结构以及数据源，能够很容易的和现有的元数据管理系统或配置管理系统结合。经过下图咱们能够清晰的看出 Storm 和 Flink 在开发统过程当中的区别。

在使用 Storm 开发时处理逻辑与实现须要固化在 Bolt 的代码。Flink 则能够经过 SQL 进行开发，代码可读性更高，逻辑的实现由开源框架来保证可靠高效，对特定场景的优化只要修改 Flink SQL 优化器功能实现便可，而不影响逻辑代码，使咱们能够把更多的精力放到数据开发中，而不是逻辑的实现。当须要离线数据和实时数据口径统一的场景时，咱们只需对离线口径的 SQL 脚本稍加改造便可，极大地提升了开发效率。同时，对比图中 Flink 和 Storm 使用的数据模型，Storm 须要经过一个 Java 的 Class 去定义数据结构，Flink Table 则能够经过元数据来定义。能够很好的和数据开发中的元数据，数据治理等系统结合，提升开发效率。

Flink使用心得

在利用 Flink-Table 构建实时数据仓库过程当中，咱们针对一些构建数据仓库的经常使用操做，好比数据指标的维度扩充，数据按主题关联，以及数据的聚合运算经过 Flink 来实现总结了一些使用心得。

1. 维度扩充

数据指标的维度扩充，咱们采用的是经过维度服务获取维度信息。虽然基于 Cellar 的维度服务一般的响应延迟能够在 1ms 如下，可是为了进一步优化 Flink 的吞吐，咱们对维度数据的关联所有采用了异步接口访问的方式，避免了使用 RPC 调用影响数据吞吐。

对于一些数据量很大的流，好比流量日志数据量在 10万秒/条这个量级，在关联 UDF 的时候内置了缓存机制，能够根据命中率和时间对缓存进行淘汰，配合用关联的 Key 值进行分区，显著减小了对外部服务的请求次数，有效的减小了处理延迟和对外部系统的压力。

2. 数据关联

数据主题合并，本质上就是多个数据源的关联，简单的来讲就是 Join 操做。Flink 的 Table 是创建在无限流这个概念上的，在进行 Join 操做时并不能像离线数据同样对两个完整的表进行关联，采用的是在窗口时间内对数据进行关联的方案，至关于从两个数据流中各自截取一段时间的数据进行 Join 操做，有点相似于离线数据经过限制分区来进行关联。同时须要注意 Flink 关联表时必须有至少一个“等于”关联条件，由于等号两边的值会用来分组。

因为 Flink 会缓存窗口内的所有数据来进行关联，缓存的数据量和关联的窗口大小成正比。所以 Flink 的关联查询，更适合处理一些能够经过业务规则限制关联数据时间范围的场景，好比关联下单用户购买以前 30 分钟内的浏览日志。过大的窗口不只会消耗更多的内存，同时会产生更大的 Checkpoint ，致使吞吐降低或 Checkpoint 超时。在实际生产中可使用 RocksDB 和启用增量保存点模式，减小 Checkpoint 过程对吞吐产生影响。

对于一些须要关联窗口期很长的场景，好比关联的数据多是几天之前的数据，对于这些历史数据，咱们能够将其理解为是一种已经固定不变的"维度"。能够将须要被关联的历史数据采用和维度数据一致的处理方法："缓存 + 离线"数据方式存储，用接口的方式进行关联。另外，须要注意 Flink 对多表关联是直接顺序连接的，所以须要注意先进行结果集小的关联。

3. 聚合运算

使用聚合运算时，Flink 对常见的聚合运算如求和、极值、均值等都有支持。美中不足的是对于 Distinct 的支持，Flink-1.6 以前的采用的方案是经过先对去重字段进行分组再聚合实现。对于须要对多个字段去重聚合的场景，只能分别计算再进行关联处理效率很低，为此咱们开发了自定义的 UDAF，实现了 MapView 精确去重、BloomFilter 非精确去重、 HyperLogLog 超低内存去重方案应对各类实时去重场景。

可是在使用自定义的 UDAF 时，须要注意 RocksDBStateBackend 模式对于较大的 Key 进行更新操做时序列化和反序列化耗时不少。能够考虑使用 FsStateBackend 模式替代。另外要注意的一点 Flink 框架在计算好比 Rank 这样的分析函数时，须要缓存每一个分组窗口下的所有数据才能进行排序，会消耗大量内存，建议在这种场景下优先转换为 TopN 的逻辑，看是否能够解决需求。

下图展现一个完整的使用 Flink 引擎生产一张实时数据表的过程：

实时数仓成果

经过使用实时数仓代替原有流程，咱们将数据生产中的各个流程抽象到实时数仓的各层当中，实现了所有实时数据应用的数据源统一，保证了应用数据指标、维度的口径的一致。在几回数据口径发生修改的场景中，咱们经过对仓库明细和汇总进行改造，在彻底不用修改应用代码的状况下就完成所有应用的口径切换。在开发过程当中经过严格的把控数据分层、主题域划分、内容组织标准规范和命名规则，使数据开发的链路更为清晰，减小了代码的耦合；再配合上使用 Flink SQL 进行开发，代码更加简洁，单个做业的代码量从平均 300+ 行的 Java 代码 ，缩减到几十行的 SQL 脚本；项目的开发时长也大幅减短，一人日开发多个实时数据指标状况也很多见。

除此之外咱们经过针对数仓各层级工做内容的不一样特色，能够进行针对性的性能优化和参数配置，好比 ODS 层主要进行数据的解析、过滤等操做，不须要 RPC 调用和聚合运算。 咱们针对数据解析过程进行优化，减小没必要要的 JSON 字段解析，并使用更高效的 JSON 包，在资源分配上，单个 CPU 只配置 1GB 的内存便可满需求。

而汇总层主要则主要进行聚合与关联运算，能够经过优化聚合算法、内外存共同运算来提升性能、减小成本；资源配置上也会分配更多的内存，避免内存溢出。经过这些优化手段，虽然相比原有流程实时数仓的生产链路更长，但数据延迟并无明显增长，同时实时数据应用所使用的计算资源也有明显减小。

展望

咱们的目标是将实时仓库建设成能够和离线仓库数据准确性，一致性媲美的数据系统，为商家，业务人员以及美团用户提供及时可靠的数据服务，同时做为到餐实时数据的统一出口，为集团其余业务部门助力。

将来咱们将更加关注在数据可靠性和实时数据指标管理，创建完善的数据监控，数据血缘检测，交叉检查机制。及时对异常数据或数据延迟进行监控和预警；同时，优化开发流程，下降开发实时数据学习成本，让更多有实时数据需求的人，能够本身动手解决问题。

参考内容

流计算框架 Flink 与 Storm 的性能对比

关于做者

伟伦， 美团到店餐饮技术部实时数据负责人，2017年加入美团，长期从事数据平台、实时数据计算、数据架构方面的开发工做。在使用 Flink 进行实时数据生产和提升生产效率上，有一些心得和产出。同时也在积极推广 Flink 在实时数据处理中的实战经验。