关于Flink:Flink-在顺丰的应用实践

本⽂由社区志愿者苗文婷整顿，内容源⾃顺丰科技大数据平台研发工程师龙逸尘在 Flink Forward Asia 2020 分享的《Flink 在顺丰的利用实际》，次要分享内容为：顺丰基于 Flink 建设实时数仓的思路，引入 Hudi On Flink 减速数仓宽表，以及实时数仓平台化建设的实际。分为以下 5 个局部：

1. 建设背景
2. 建设思路
3. 落地实际
4. 利用案例
5. 将来布局

一、建设背景

顺丰是国内当先的快递物流综合服务商，通过多年的倒退，顺丰应用大数据技术支持高质量的物流服务。以下是一票快件的流转过程，能够看到从客户下单到最终客户收件的整个过程是十分长的，其中波及的一些解决逻辑也比较复杂。为了应答简单业务的挑战，顺丰进行了数据仓库的摸索。

传统数仓次要分为离线和实时两个局部。

• 离线局部以固定的计算逻辑，通过定时调度，实现数据抽取，荡涤，计算，最初产出报表；
• 而实时局部则是需要驱动的，用户须要什么，就马上着手开发。

这种数仓架构在数据量小、对实时性要求不高的状况下运行得很好。然而随着业务的倒退，数据规模的扩充和实时需要的一直增长，传统数仓的毛病也被放大了。

• 从业务指标的开发效率来看

实时指标采纳的是需要驱动的、纵向烟囱式的开发模式，须要用户手写 Flink 工作进行开发，这种开发方式效率低门槛高，输入的指标很难对立治理与复用。

• 从技术架构方面来看

离线和实时两套架构是不对立的，开发方式、运维形式、元数据方面都存在差别。传统架构整体还是以离线为主，实时为辅，依赖离线 T+1 调度导出报表，这些调度工作通常都运行在凌晨，导致凌晨时集群压力激增，可能会导致报表的产出不稳固；如果重要的报表产出有提早，相应的上游的报表产出也会呈现提早。这种以离线为主的架构无奈满足精细化、实时化经营的须要。

• 从平台治理的角度来看

传统数仓的实时指标开发是比拟粗放的，没有 Schema 的标准，没有元数据的治理，也没有买通实时和离线数据之间的分割。

为了解决传统数仓的问题，顺丰开始了实时数仓的摸索。实时数仓和离线数仓实际上解决的都是雷同的业务问题，最大的区别就在于时效性。

• 离线数仓有小时级或天级的提早；
• 而实时数仓则是秒级或分钟级的提早。

其余个性，比方数据源、数据存储以及开发方式都是比拟相近的。因而，咱们心愿：

• 用户能从传统数仓平滑迁徙到实时数仓，保持良好的体验；
• 同时对立实时和离线架构，放慢数据产出，缩小开发的撕裂感；
• 增强平台治理，升高用户应用门槛，进步开发效率也是咱们的指标。

二、建设思路

通过总结，咱们提炼出以下 3 个实时数仓的建设思路。首先是通过对立数仓规范、元数据以及开发流程，使得用户达到开发体验上的批流对立。随后，引入 Hudi 减速数仓宽表，基于 Flink SQL 建设咱们的实时数仓。最初是增强平台治理，进行数仓平台化建设，实现数据对立接入、对立开发、以及对立的元数据管理。

1. 批流对立的实时数仓

建设批流对立的实时数仓能够分为以下 3 个阶段：

1.1 对立数仓标准

首先，无规矩不成方圆，建设数仓必须有对立的数仓标准。对立的数仓标准包含以下几个局部：

• 设计规范
• 命名标准
• 模型标准
• 开发标准
• 存储标准
• 流程标准

对立好数仓标准之后，开始数仓层级的划分，将实时和离线统一规划数仓层级，分为 ODS、DWD、DWS、ADS 层。

1.2 对立元数据

基于以上对立的数仓标准和层级划分模型，能够将实时和离线的元数据进行对立治理。上游的数据治理过程，比方数据字典、数据血统、数据品质、权限治理等都能够达到对立。这种对立能够积淀实时数仓的建设成绩，使数仓能更好的落地施行。

1.3 基于 SQL 对立开发流程

开发人员都晓得，应用 DataStream API 开发 Flink 工作是比较复杂的。在数据量比拟大的状况下，如果用户应用 API 不标准或者开发能力有余，可能会导致性能和稳定性的问题。如果咱们能将实时开发的过程对立到 SQL 上，就能够达到缩小用户开发成本、学习老本以及运维老本的目标。

之前提到过咱们曾经对立了实时和离线的元数据，那么就能够将上图右边的异构数据源和数据存储形象成对立的 Table，而后应用 SQL 进行对立的数仓开发，也就是将离线批处理、实时流解决以及 OLAP 查问对立 SQL 化。

1.4 实时数仓计划比照

实现了数仓标准、元数据、开发流程的对立之后，咱们开始摸索数仓架构的具体架构计划。业界目前的支流是 Lambda 架构和 Kappa 架构。

• Lambda 架构

Lambda 架构是在原有离线数仓的根底上，将对实时性要求比拟高的局部剥离进去，减少了一个实时速度层。Lambda 架构的毛病是须要保护实时和离线两套架构和两套开发逻辑，保护老本比拟高，另外两套架构带来的资源耗费也是比拟大的。

• Kappa 架构

为了应答 Lambda 架构的缺点，Jay Kreps 提出了 Kappa 架构，Kappa 架构移除了原有的离线局部，应用纯流式引擎开发。Kappa 架构的最大问题是，流数据重放解决时的吞吐能力达不到批处理的级别，导致重放时产生肯定的延时。

• 实时数仓计划比照与理论需要

在实在的生产实践中，并不是肯定要严格遵循标准的 Lambda 架构或 Kappa 架构，能够是两者的混合。比方大部分指标应用流式引擎开发，少部分重要的指标应用批处理开发，并减少数据校对的过程。

在顺丰的业务场景中，并非所有用户都须要纯实时的表，许多用户的报表还是依赖离线 T+1 调度产出的宽表，如果咱们可能减速宽表的产出，那么其余报表的时效性也能相应地失去进步。

另外，这个离线 T+1 调度产出的宽表，须要聚合 45 天内多个数据源的全量数据，不论是 Lambda 架构还是 Kappa 架构，都须要对数据进行全量聚合，如果可能间接更新宽表，就能够防止全量从新计算，大大降低资源耗费和延时。

2. 引入 Hudi 减速宽表

之前说过，保护 Lambda 架构的复杂性在于须要同时保护实时和离线两套零碎架构。而对于这个毛病，咱们能够通过批流对立来克服。

通过衡量，咱们决定革新原有 Lambda 架构，通过减速它的离线局部来建设数仓宽表。此时，就须要一个工具来实时疾速的更新和删除 Hive 表，反对 ACID 个性，反对历史数据的重放。基于这样的需要，咱们调研了市面上的三款开源组件：Delta Lake、Iceberg、Hudi，最初抉择 Hudi 来减速宽表。

2.1 Hudi 要害个性

Hudi 的要害个性包含：可回溯历史数据，反对在大规模数据集中依据主键更新删除数据；反对数据增量生产；反对 HDFS 小文件压缩。这些个性恰好能满足咱们的需要。

2.2 引入 Hudi 减速宽表

引入 Hudi 有两种形式减速数仓。首先，在 ODS 层引入 Hudi 实现实时数据接入，将 ODS 层 T+1 的全量数据抽取改成 T+0 的实时接入，从数据源头实现 Hive 表的减速。

另外，应用 Flink 生产 Kafka 中接入的数据，进行荡涤聚合，通过 Hudi 增量更新 DWD 层的 Hive 宽表，将宽表从离线减速成准实时。

2.3 构建实时数仓宽示意例

这里通过一个例子介绍如何构建实时数仓宽表。

假如运单宽表由运单表，订单表和用户表组成，别离蕴含运单号、运单状态、订单号、订单状态、用户 ID、用户名等字段。

首先将运单表数据插入宽表，运单号作为宽表主键，并且将运单号和订单号的映射存入长期表。当订单表数据更新后，首先关联用户维表，获取用户名，再从长期表中获取对应运单号。最初依据运单号将订单表数据增量插入宽表，以更新宽表状态。

3. 最终架构

引入 Hudi 后，基于 Lambda 架构，咱们定制化的实时数仓最终架构如下图所示。实时速度层通过 CDC 接入数据到 Kafka，采纳 Flink SQL 解决 Kafka 中的数据，并将 ODS 层 Kafka 数据荡涤计算后通过 Hudi 准实时更新 DWD 层的宽表，以减速宽表的产出。离线层采纳 Hive 存储及解决。最初由 ADS 层提供对立的数据存储与服务。

除了制订数仓规范和构建数仓架构，咱们还须要构建数仓平台来束缚开发标准和流程，晋升开发效率，进步用户体验。

站在数据开发人员的角度，咱们不仅要提供疾速的数据接入能力，还须要关注开发效率以及对立的元数据治理。因而能够基于 Table 和 SQL 形象，对数据接入、数据开发、元数据管理这三个次要性能进行平台化，为实时数仓用户提供对立、便捷、高效的体验。

三、落地实际

1. Hudi On Flink

顺丰是最早将 Hudi On Flink 引入生产实践的公司，顺丰外部应用版本基于 T3 出行的外部分支进行了许多批改和欠缺，大大晋升了 Hudi on Flink 的性能和稳定性。

1.1 实现原理

这里介绍下 Hudi On Flink 的原理。Hudi 原先与 Spark 强绑定，它的写操作实质上是批处理的过程。为理解耦 Spark 并且对立 API，Hudi On Flink 采纳的是在 Checkpoint 期间攒批的机制，在 Checkpoint 触发时将这一批数据 Upsert 到 Hive，依据 Upsert 后果对立提交或回滚。

Hudi On Flink 的实现流能够合成为几个步骤：

1. 首先应用 Flink 生产 Kafka 中的 Binlog 类型数据，将其转化为 Hudi Record。
2. Hudi Record 进入 InstantTime Generator，该 Operator 并不对数据做任何解决，只负责转发数据。它的作用是每次 Checkpoint 时在 Hudi 的 Timeline 上生成全局惟一且递增的 Instant，并下发。
3. 随后，数据进入 Partitioner，依据分区门路以及主键进行二级分区。分区后数据进入 File Indexer，依据主键找到在 HDFS 上须要更新的对应文件，将这个对应关系按文件 id 进行分桶，并下发到上游的 WriteProcessOperator。
4. WriteProcessOperator 在 Checkpoint 期间会积攒一批数据，当 Checkpoint 触发时，通过 Hudi 的 Client 将这批数据 Upsert 到 HDFS 中，并且将 Upsert 的后果下发到上游的 CommitSink。
5. CommitSink 会收集上游所有算子的 upsert 后果，如果胜利的个数和上游算子的并行度相等时，就认为本次 commit 胜利，并将 Instant 的状态设置为 success，否则就认为本次 commit 失败并进行回滚。

1.2 优化

顺丰基于社区代码对 Hudi On Flink 进行了一些优化，次要目标是加强性能和晋升稳定性。

• 二级分区

对于增量写入的场景，大部分的数据都写入当天的分区，可能会导致数据歪斜。因而，咱们应用分区门路和主键 id 实现二级分区，防止攒批过程中单个分区数据过多，解决数据歪斜问题。

• 文件索引

Hudi 写入过程的瓶颈在于如何疾速找到记录要写入的文件并更新。为此 Hudi 提供了一套索引机制，该机制会将一个记录的键 + 分区门路的组合映射到一个文件 ID. 这个映射关系一旦记录被写入文件组就不会再扭转。Hudi 以后提供了 HBase、Bloom Filter 和内存索引 3 种索引机制。然而通过生产实践，HBase 索引须要依赖内部的组件，内存索引可能存在 OOM 的问题，Bloom Filter 存在肯定的误算率。咱们钻研发现，在 Hudi 写入的 parquet 文件中存在一个暗藏的列，通过读取这个列能够拿到文件中所有数据的主键，因而能够通过文件索引获取到数据须要写入的文件门路，并保留到 Flink 算子的 state 中，也防止了内部依赖和 OOM 的问题。

• 索引写入拆散

原先 Hudi 的 Upsert 过程，写入和索引的过程是在一个算子中的，算子的并行度只由分区门路来决定。咱们将索引和写入的过程进行拆散，这样能够进步 Upsert 算子的并行度，进步写入的吞吐量。

• 故障复原

最初咱们将整个流程的状态保留到 Flink State 中，设计了一套基于 State 的故障复原机制，能够保障端到端的 exactly-once 语义。

2. 实时数仓的产品化

在实时数仓产品化方面，咱们也做了一些工作。提供了包含数据接入、元数据管理、数据处理在内的数仓开发套件。

2.1 实时数据接入

实时数据接入采纳的是表单式的流程接入形式，屏蔽了简单的底层技术，用户只须要通过简略的操作就能够将内部数据源接入到数仓体系。以 MySQL 为例，用户只须要抉择 MySQL 数据源，平台就会主动抽取并展现 Schema，用户确认 Schema 之后，就会将 Schema 插入到平台元数据中。

随后，用户抉择有权限的集群，设置 Hive 表的主键 ID 和分区字段，提交申请之后，平台就会主动生成 Flink 工作，抽取数据到 Kafka 并主动落入 Hive 表中。对数据库类型的数据源，还反对分库分表性能，将分库分表的业务数据写入 ODS 层的同一张表。另外也反对采集主从同步的数据库，从从库中查问存量数据，主库拉取 Binlog，在加重主库压力的同时升高数据同步提早。

2.2 实时元数据更新

实时元数据更新的过程，还是以 MySQL 为例。CDC Source 会抽取数据库中的 Binlog，辨别 DDL 和 DML 语句别离解决，DDL 语句会上报到元数据中心，DML 语句通过转化变成 avro 格局的 Binlog 数据发送到 Kafka，如果上游有写入到 Hive 的需要，就生产 Kafka 的数据通过 Hudi Sink 写入到 Hive。

2.3 数据资产管理体系

基于实时数据的对立接入，并将其与现有的离线数仓联合，咱们构建了数据资产管理体系。包含标准数仓规范，对立治理元数据，晋升数据品质，保障数据安全，盘点数据资产。

3. 实时计算平台架构

有了数据对立接入的根底和数据资产资产管理体系的保驾护航，咱们还须要一个数据开发套件，将整个数据开发的过程整合到实时计算平台。实时计算平台的最底层是数据接入层，反对 Kafka 和 Binlog 等数据源。上一层是数据存储层，提供了 Kafka、ES、HBase、Hive、ClickHouse、MySQL 等存储组件。反对 JStorm、Spark Streaming、Flink 计算引擎。并进行了框架封装和公共组件打包。

3.1 多种开发模式 – JAR & DRAG

实时计算平台提供了多种开发模式供不同用户抉择。以 Flink 为例，Flink JAR 模式由用户编写 Flink 工作代码，打成 jar 包上传到平台，满足高级用户的需要。Flink DRAG 模式则是图形化的拖拽式开发，由平台封装好公共组件之后，用户只须要拖拽公共组件，将其组装成一个 Flink 工作，提交至集群运行。

3.2 多种开发模式 – SQL

实时计算平台同样提供 SQL 开发模式，反对手动建表，依据元数据自动识别表及设置表属性。反对创立 UDF、自动识别 UDF、执行 DML 等。

3.3 工作管控

在工作管控方面，实时计算平台尽量简化工作的配置，屏蔽了一些简单的配置。用户开发实现之后，只须要抉择集群，填写资源，就能将工作提交到集群中运行。对每个工作，平台还提供了历史版本控制能力。

当用户操作工作时，平台会主动解析工作的配置，依据不同的组件提供不同的选项。比方抉择了 Kafka 数据源，启动的时候，能够抉择从上次生产地位、最早地位、最新地位或指定地位启动。

工作复原方面，用户能够抉择从 Savepoint 启动已进行的 Flink 工作，便于疾速复原历史状态。

3.4 工作运维

对实时工作来说，工作运维是一个难点也是一个痛点。平台提供了日志查问性能，采集历史的启动日志和工作运行日志，用户能够不便的进行比照和查问。

当工作启动之后，平台会主动采集并上报工作的指标，用户能够依据这些指标自定义告警配置，当告警规定被触发时，平台会通过各种形式告警到用户。最初，平台提供了指标的实时监控看板，当然用户也能够自行在 Grafana 中配置监控看板。

通过采集日志、指标以及监控告警，以及过往的历史教训，咱们实现了一个智能的机器客服，能够实现工作故障的一些自助诊断。这些动作大大降低了工作的运维老本，加重平台研发人员的压力。

3.5 Flink 工作稳定性保障

实时作业运维最关注的是稳定性，在保障 Flink 工作稳定性上咱们也有一些实际。首先提供多种异样检测和监控告警的性能，不便用户疾速的发现问题。每个工作都会定时的生成工作快照，保留工作历史的 Savepoint，以不便工作回滚和故障复原。工作可能会因为某种异样起因导致工作失败，工作失败之后会被平台从新拉新，并指定从上次失败的地位开始从新生产。

基于 Zookeeper 的高可用机制，以保障 JobManager 的可用性。反对多集群、多机房的容灾切换能力，能够将工作一键切换至容灾集群上运行。实现了一套实时离线集群隔离、队列治理的资源隔离零碎。

四、利用案例

以业务宽表计算为例，须要获取 45 天内的多个数据源的数据，进行计算聚合。如果应用离线数仓，大略须要 3000 核的 CPU、12000G 的内存，耗时 120 ～ 150 min 实现计算，解决的数据量大略为 450T。如果应用实时数仓，大略须要 2500 核的 CPU、1400G 的内存，更新宽表大略有 2～5 min 的延时，解决的数据量约为 18T。

五、将来布局

顺丰的实时数仓建设获得了一些成绩，但将来仍须要进行一直的优化。

1. 加强 SQL 能力

首先，心愿可能反对更多 SQL 的语法和个性，反对更多可用的连接器，以及实现 SQL 工作的主动调优等。

2. 精细化资源管理

其次，基于 Flink On Kubernets、工作的主动弹性扩缩容，Task 级别的细粒度资源调度实现精细化的资源调度治理，使得 Flink 工作达到全面的弹性化和云原生化。

3. 流批一体

最初，心愿可能实现流批一体，通过对立的高度兼容性的 SQL，通过 SQL 解析以及引擎的适配，通过 Flink 对立的引擎去解决流和批。