关于sql:Flink-流批一体在小米的实践

简介： 基于流批一体的革新，无论是实时还是离线都只须要保护一套计算框架，为业务开发人员、平台提供方和计算引擎的反对方节俭了大量人力资源。

摘要：本文整顿自小米软件开发工程师金风在 Flink Forward Asia 2021 流批一体专场的演讲。本篇内容次要分为三个局部：

1. 小米的大数据倒退演变
2. 流批一体的平台建设
3. 流批一体利用场景
4. 将来布局

点击查看直播回放 & 演讲 PDF

一、小米的大数据倒退演变

• 2019 年之前，小米的实时计算次要以 SparkStreaming 为主，少部分 Storm，离线计算以 Spark 为主。
• 2019 年，开始接入 Flink，并广泛应用于信息流搜寻举荐、广告实时样本、实时 ETL 等场景，逐渐替换了原来的 SparkStreaming 作业，得益于 Flink 框架的各种优良个性，咱们在作业的正确性，实时性，资源应用效率方面都有较大晋升。
• 2020 年，开始接入应用 FlinkSQL，并宽泛用于实时数仓的建设和实时 ETL 作业的开发。FlinkSQL 的实时数仓将数据链路由 T+1 升高到了秒级。
• 2021 年，开始接入数据湖 Iceberg，基于 Flink 和 Iceberg 来构建流批一体的实时数仓解决方案，并在小米外部的局部业务进行了落地，证实流批一体在赋能业务、晋升作业开发效率、简化链路节俭资源的方面是可行的。

上图是小米以后的实时和离线框架，目前是多种框架并存的状态。业务开发人员无论是写 SQL 作业还是写 Jar 包作业，都至多要保护两套代码。公司外部的计算引擎团队也须要花两拨人力别离去保护不同的计算框架，同时平台层也须要对不同的计算引擎去做不同的适配。

基于流批一体的革新，无论是实时还是离线都只须要保护一套计算框架，为业务开发人员、平台提供方和计算引擎的反对方节俭了一半的人力资源。

二、流批一体的平台建设

为了摸索流批一体，咱们也做了很多相干的摸索和实际。

对于流批一体的平台化建设，次要分为 4 个方面，别离是元数据管理、权限治理、作业调度以及 Flink 的生态建设。

2.1 元数据管理

咱们基于 Metacat 做了对立的元数据管理，由 Metacat 对立对接上游不同的存储系统和上游的计算引擎。

基于 Metacat，外部的所有零碎都被对立划分成三级构造，与 FlinkSQL 的三级构造绝对应。

• 第一级 Catalog，次要由服务名和集群名拼接而成。
• 第二级 Database，它与大部分零碎的 Database 保持一致。没有 Database 的零碎默认应用 default 来代替。
• 第三级 Table，也与零碎的 Table 保持一致，比方音讯队列的 topic 名，Elasticsearch 的索引名。

在构建好对立的元数据管理之后，只须要写一条 DML 语句即可实现一个实时将音讯队列数据入湖作业的开发。

2.2 权限治理

在实现了对立的元数据管理后，在开发 Flink SQL 作业时，所有零碎都被形象为一个三级的构造表，业务能够应用三级表名援用任意一个零碎的表。同时咱们基于 Ranger 也做了对立的权限治理，在 SQL 层对立治理所有的资源权限。

咱们在计算引擎层做了对立的权限治理，同时笼罩了 Flink SQL 和 Flink Jar。Flink SQL 作业能够在生成物理执行打算时获取到 SQL 援用的的 Source 和 Sink 表，以及 select 的 Source 表的字段名。基于以上信息，即可实现字段级别的鉴权。同时咱们为 Flink Jar 用户提供了对立的工具类，同时也对接了 Flink Catalog，因而能够做到 Jar 包作业的权限校验。

如上图所示，对立治理了元数据和权限之后，业务开发人员在开发 Flink SQL 作业时就能够十分不便地抉择不同零碎的表，包含 Doris、Kudu、Hive 等，作业由后端对立进行提交并鉴权。在作业提交时，咱们也能十分不便得获取到作业的血统。

2.3 作业调度

在作业调度方面小米也做了一些尝试。如上图右边这段 SQL，在离线调度模式下它是一个批作业，但在实时调度下它就是一个流作业。在批流混合的调度下，会先启动批作业，执行实现之后再启动流作业。

批流混合对于调度器来说是实时的作业。咱们次要的改变是在 Flink SQL 的模板作业中先启动一个 SQL 的批作业，执行实现之后再启动 Flink SQL 的实时作业。

2.4 Flink 的生态建设

Flink 插件化的 connector 设计能够十分不便地拓展不同的 connector。无论是 Flink 官网还是其余社区，都提供了十分多的 connector 反对。小米外部也实现了很多品种 connector，只有欠缺了 Flink 的生态建设，它跨平台设计的计算能力能力真正体现进去。

对于 Iceberg connector，社区曾经实现了批量读写和流式入湖的相干性能。另外流式生产也是一项比拟重要的性能，如果不反对流式生产，在数仓的链路中就只能革新 ODS 层，上游链路只能以批的形式来解决，无奈做到全链路真正的实时处理。因而反对 Iceberg 的增量生产是实时链路中必不可少的一环。

对于流批一体生态比拟重要的还有 Hybrid Source 和 CDC Sink。

Hybrid Source 在社区曾经有相干的实现，它可能将两种不同的 Source 进行组合，大多数组合是无限流 + 有限流，这样就能做到批流混合。

小米在平台层曾经对立治理了所有零碎的表，因而在实现 Hybrid Source 时就无需填写对应表的构造信息和比拟繁琐的参数信息，只须要依照程序将须要读的表名配置在参数中即可，Hybrid Source 会依照配置的程序，顺次读取须要的表。此外，还能够进行不同零碎的组合，比拟罕用的是 MySQL 和音讯队列的组合，先全量生产 MySQL 中的数据，再增量生产音讯队列的数据。

CDC Sink 次要配合 Hybrid Source 来应用。CDC Sink 同样也对接了外部的 Catalog，它对立治理了 Schema 的变更操作。数据达到上游的 connector 时不须要再去解决繁琐的 Schema 变更逻辑，只须要将实在的数据以实在的 Schema 写入对应的零碎即可。

无论是 Hybrid Source 还是 CDC Sink，在 Flink 框架层的字段类型都有一个 barrier 字段，它能够封装任意构造的数据，也能够做 Schema 变更。然而一些字段类型不匹配的状况，只有在运行时才会裸露进去。

三、流批一体利用场景

大多公司都有数据导入和导出的需要，基于 Flink 丰盛的生态，咱们能够十分不便地实现不同场景的数据集成，次要包含离线集成、实时集成以及批流混合数据集成。

首先是离线的数据集成。咱们应用 Flink SQL Batch 作业替换了之前的 Data X，借助 Flink 的生态，能够十分不便地实现不同零碎数据导入导出的需要，也取得了更丰盛的 Source Sink 生态。同时还基于 Flink SQL 能够十分不便地实现字段的映射，同时 Flink SQL 作为分布式框架，能够很不便提供并发导数的需要。

其次是实时数据集成，次要分为两个局部：

• 第一局部是实时数据的收集，小米外部次要分为两大类，别离是日志数据和 DB 的 Binlog 数据。这里次要介绍 DB 零碎的 Binlog 数据收集。最后咱们应用小米自研的 LCS Binlog 服务来进行对立的 Binlog 收集，相似于 Canal 服务，通过该服务将 Binlog 的数据对立收集到音讯队列中。
• 第二局部则是数据的转储，将应用 Spark Streaming 工作将音讯队列中的数据导入其余零碎，比方 Kudu 或 HDFS。

当初咱们应用 Flink 对 Binlog 的收集和转储链路都进行了革新。应用 Flink CDC 收集 Binlog 数据，并写入音讯队列中。同时通过 Flink 将音讯队列的数据转储到其余零碎，比方 Kudu、Doris、Iceberg 等等。

在理论的应用中往往须要用流批混合的形式，以实用不同的场景，比方分库分表场景，局部链路重做场景，新增库表等场景。以后，应用 Flink CDC 工作来收集库级别的 Binlog 数据，如果依照表级别来进行收集，会对 MySQL 服务造成较大的压力。将数据收集到音讯队列后，再针对不同的收集场景，起不同的作业来进行转储。

Flink CDC 收集的数据是库级别，当某张表的数据须要重做时，无奈将库级别的数据重做，因为这样会影响到其余的表。所以，对于单表的全量数据，咱们会间接从 MySQL 读取，再从音讯队列中读取增量的数据，因而这里须要用 Hybrid Source 别离读取 MySQL 和音讯队列中的数据。

另一种批流混合的数据集成场景则是批作业和流作业混合应用。

在反对 TiDB 的数据收集和转储时，咱们无奈应用 Hybrid Source，因为 TiDB 的全量数据往往十分大，咱们须要起大量并发可能减速全量数据的转储，而增量数据则只须要较小并发即可。在全量数据局部咱们应用 Flink SQL Batch 作业来实现，能够灵便调整并发，且绝对于实时作业处理效率更高，增量局部则以较小的并发能转储即可。

另外一个比拟重要的业务场景是实时数仓的构建。在小米外部也经验了从传统离线数仓到 Lambda 架构再到以后基于数据湖的实时实仓。这三种场景有不同的优缺点，会利用于不同的业务场景。有两个比拟典型的案例，别离是小米手机激活统计以及小米销售服务实时数仓。

上图是小米手机激活业务流程，首先是激活数据的收集，通过不同的渠道来收集日志，并进行对立的汇总和荡涤。通过采集数据配合维表 join 可能检测到提前激活的 case，同时也可能基于一些维度数据进行数据荡涤，判断出哪些属于天然激活，哪些属于失常的沉闷日志。

小米手机激活数仓的整体架构波及到实时链路和离线链路，这里次要介绍实时链路。咱们采纳的维度表次要是 HBase 和 FileSystem，HBase 用于保留历史全量的惟一 ID，Hive 次要保留大量的维度数据，最终的后果会实时落到 Kudu 中，业务就能够通过 OLAP 引擎查到实时的激活数据。同时离线链路也是必不可少的，实时和离线产生的数据整体的重合率达到 99.94%。

上述链路中最要害的点是须要应用 HBase 来保留历史的全量 ID 来去重。这其中 join 历史全量数据 HBase 表是最要害的中央，最开始咱们应用同步的 lookup join 形式，然而遇到了较大的性能瓶颈，后改用异步 join 形式，最终整体的处理速度有了数十倍的晋升。

小米的销售服务波及到多个模块，包含订单、物流、商品、售后门店，在构建过程中咱们也遇到了十分多的问题，最终证实基于 Flink SQL 革新离线链路构建实时数仓的计划是可行的。

上图是销售服务数仓的整体架构，销售服务用到的维度表次要来自于音讯队列和 FileSystem。在销售服务的场景下，无论是订单表还是商品类目表都会实时更新。进行关联时，无论哪一条流有更新，后果都须要更新。因而销售的服务数仓大多采纳双流 join，而双流 join 随同而来的就是状态问题。

在 Flink SQL 中咱们无奈精确管制某个算子的状态过期的策略，因而只能设置一个对立的状态过期工夫，如果一段时间内某些状态没有被拜访则会被清理，但这个场景是有局限性的，针对于物流售后这些场景，单条记录在整个实时流中的周期可能会超过一个月，但个别状况下咱们无奈将 Flink 作业的状态超时工夫设置为一个月，产生的状态量太多会导致解决效率变低，也不利于作业链路的回溯。一旦作业呈现问题，上游的数据都须要重做，音讯队列中大多数据保留不会超过 7 天。

在销售服务的某些场景中中，咱们引入了离线作业，依据后果表的数据，获取状态还未完结的数据，将对应的维度数据写回到实时流中，确保这些维度数据不会过期，当主表的数据达到 join 算子，就能够失去正确的数据，即便主表的某条记录超过一个月才呈现变更。

上图是小米 APP 近实时数仓的架构。通过采集日志模块将日志对立收集到音讯队列中。因为是日志数据，只须要应用到 Iceberg v1 表，两头的 DWD 和 DM 层都应用 Flink SQL 进行实时生产和解决，而后写入到 Iceberg v2 表中。

对于有些须要保留历史的全量数据并保证数据准确性，但对时效性要求不高的数据表，咱们采纳离线的解决思路，基于 DWD 层进行离线解决。每天应用 t-1 的数据去修改 t-2 的数据，通过一直修改历史数据，能够很大水平地保障这些表的准确率。

对于一些旧的链路或者上游数据由其余业务方提供, 且短期内无奈批改且无奈产生 CDC 数据的零碎，咱们采纳了 Spark Merge Into 作业定时调度来产生增量的数据，并实时写入到 Iceberg v2 表中。在生产实践中，通过 Merge Into 产生的数据提早大略是 5~8 分钟，前面链路的提早都能够管制在分钟级以内，因而全链路的提早根本能够管制在 10 分钟之内，相比之前 t+1 的提早有了微小的晋升。

对于可能从源头产生 CDC 数据的零碎，咱们会将实时数据写入到音讯队列中，而后实时入湖。整体架构如上图所示，实时局部次要用应用 Flink 和音讯队列的组合来达到秒级的提早，再将音讯队列的数据 Sink 到 Iceberg 中，用于 Query 查问。同时上游有一条离线链路一直通过 Merge Into 进行修改，保障 Iceberg 中后果的正确性。

整个链路基于音讯队列保障了全链路的时效性，同时基于数据湖保障了 Query 的查问时效，另外通过离线的 Merge Into 一直地修改，保障了最终后果的准确性。

尽管数仓的架构整体都根本相似，然而针对不同业务场景及不同的需要，理论的链路复杂度还是有较大的区别。因而在应用 Flink 构建数仓时，肯定要结合实际的需要抉择正当的计划。

四、将来布局

以后 Flink + Iceberg 的数据湖解决方案在小米曾经初步落地，将来咱们能够晋升的空间仍然十分大，咱们会一直跟进社区，持续推动小米外部流批一体化的建设。

后续咱们会将 Flink SQL Batch 用于更加简单的场景。以后 Flink SQL Batch 施展的场景无限，次要使用于批量导出的场景，置信将来它会施展更大的价值。

其次，咱们会跟进社区的 built in dynamic table [1]，联合音讯队列和数据湖，兼顾时效性和准确性，晋升用户的体验。

同时咱们也会降级 Hybrid Source connector，以后社区的 Hybrid Source 是适配最新版的 Source 接口，新版的 Source 接口在对接其余零碎时灵便度更高。

[1] https://cwiki.apache.org/confluence/display/Flink/FLIP-188%3A+Introduce+Built-in+Dynamic+Table+Storage

点击查看直播回放 & 演讲 PDF

版权申明： 本文内容由阿里云实名注册用户自发奉献，版权归原作者所有，阿里云开发者社区不领有其著作权，亦不承当相应法律责任。具体规定请查看《阿里云开发者社区用户服务协定》和《阿里云开发者社区知识产权爱护指引》。如果您发现本社区中有涉嫌剽窃的内容，填写侵权投诉表单进行举报，一经查实，本社区将立即删除涉嫌侵权内容。