本文整理自哔哩哔哩基础架构部资深研发工程师张杨在 Flink Forward Asia 2021 平台建设专场的演讲。主要内容包括：

平台建设增量化AI On Flink

在过去的一年里，B站围绕 Flink 主要做了三个方面的工作：平台建设、增量化和 AI on Flink。实时平台是实时业务的技术底座，也是 Flink 面向用户的窗口，需要坚持持续迭代优化，不断增强功能，提升用户效率。增量化是我们在增量化数仓和流批一体上的尝试，在实时和离线之间找到一个更好的平衡，加速数仓效率，解决计算口径问题。AI 方向，我们也正在结合业务做进一步的探索，与 AIFlow 社区进行合作，完善优化机器学习工作流。

一、 平台建设1.1 基础功能完善

在平台的基础功能方面，我们做了很多新的功能和优化。其中两个重点的是支持 Kafka 的动态 sink 和任务提交引擎的优化。

我们遇到了大量这样的 ETL 场景，业务的原始实时数据流是一条较大的混合数据流，包含了数个子业务数据。数据通过 Kafka 传输，末端的每个子业务都对应单独的处理逻辑，每个子业务都去消费全量数据，再进行过滤，这样的资源消耗对业务来说是难以接受的，Kafka 的 IO 压力也很大。因此我们会开发一个 Flink 任务，对混合数据流按照子业务进行拆分，写到子业务对应的 topic 里，让业务使用。

技术实现上，早期 Flink SQL 的写法就是写一个 source 再写多个 sink，每个 sink 对应一个业务的 topic，这确实可以满足短期的业务诉求，但存在的问题也较多：

第一是数据的倾斜，不同的子业务数据量不同，数据拆分后，不同 sink 之处理的数据量也存在较大差别，而且 sink 都是独立的 Kafka producer，高峰期间会造成 sink 之间资源的争抢，对性能会有明显的影响;

第二是无法动态增减 sink，需要改变 Flink SQL 代码，然后重启任务才能完成增减 sink。过程中，不仅所有下游任务都会抖动，还有一个严重的问题就是无法从 savepoint 恢复，也就意味着数据的一致性无法保证;

第三是维护成本高，部分业务存在上百个子分流需求，会导致 SQL 太长，维护成本极高。

基于以上原因，我们开发了一套 Kafka 动态 sink 的功能，支持在一个 Kafka sink 里面动态地写多个 topic 数据，架构如上图。我们对 Kafka 表的 DDL 定义进行了扩展，在 topic 属性里支持了 UDF 功能，它会根据入仓的数据计算出这条数据应该写入哪个 Kafka 集群和 topic。sink 收到数据后会先调用 UDF 进行计算，拿到结果后再进行目标集群和 topic 数据的写入，这样业务就不需要在 SQL 里编写多个 sink，代码很干净，也易于维护，并且这个 sink 被所有 topic 共用，不会产生倾斜问题。UDF 直接面向业务系统，分流规则也会平台化，业务方配置好规则后，分流实施自动生效，任务不需要做重启。而且为了避免 UDF 的性能问题，避免用户自己去开发 UDF，我们提供了一套标准的分流，做了大量的缓存优化，只要按照规范定义好分流，规则的业务表就可以直接使用 UDF。

目前内部几个千亿级别的分流场景，都在这套方案下高效运行中。

基础功能上做的第二个优化就是任务的提交引擎优化。做提交器的优化主要是因为存在以下几个问题：

第一，本地编译问题。Flink SQL 任务在 Yarn 上的部署有三种模式：per-job、application 和 yarn-session。早前我们一直沿用 per-job 模式，但是随着任务规模变大，这个模式出现了很多的问题。per-job 模式下，任务的编译是在本地进行再提交到远程 app master，编译消耗提交引擎的服务性能，在短时批量操作时很容易导致性能不足;

第二，多版本的支持问题。我们支持多个 Flink 版本，因此在版本与提交引擎耦合的情况下，需要维护多个不同代码版本的提交引擎，维护成本高;

第三，UDF 的加载。我们一直使用 Flink 命令里的 -c 命令进行 UDF 传递，UDF 代码包存在 UDFS 上，通过 Hadoop 的 web HDFS 协议进行 cluster 加载，一些大的任务启动时，web HDFS 的 HTTP 端口压力会瞬间增大，存在很大的稳定隐患;

第四，代码包的传输效率。用户代码包或者 Flink 引擎代码包都要做多次的上传下载操作，遇到 HDFS 反应较慢的场景,耗时较长，而实时任务希望做到极致的快速上下线。

因此我们做了提交器的优化：

首先引入了 1.11 版本以上支持的 application 模式，这个模式与 per-job 最大的区别就是 Flink 任务的编译全部移到了 APP master 里做，这样就解决了提交引擎的瓶颈问题;

在多版本的支持上面，我们对提交引擎也做了改造，把提交器与 Flink 的代码彻底解耦，所有依赖 Flink 代码的操作全部抽象了标准的接口放到了 Flink 源码侧，并在 Flink 源码侧增加了一个模块，这个模块会随着 Flink 的版本一起升级提交引擎，对通用接口的调用全部进行反射和缓存，在性能上也是可接受的;

而且 Flink 的多版本源码全部按照 maled 模式进行管理，存放在 HDFS。按照业务指定的任务版本，提交引擎会从远程下载 Flink 相关的版本包缓存到本地，所以只需要维护一套提交器的引擎。Flink 任何变更完全和引擎无关，升级版本提交引擎也不需要参与;

完成 application 模式升级后，我们对 UDF 和其他资源包的上传下载机制也进行了修改，通过 HDFS 远程直接分发到 GM/TM 上，减少了上传下载次数，同时也避免了 cluster 的远程加载。

1.2 新任务构建模式

平台之前支持 Flink 的构建模式主要有两种， SQL 和 JAR 包。两者的优劣势都很明显，SQL 简单易用门槛低，但是不够灵活，比如一些定时操作在 SQL 里面无法进行。JAR 包功能完善也灵活，但是门槛高，需要学习 Flink datastream 一整套 API 的概念，非开发人员难以掌握，而我们大量的用户是数仓，这种JAR包的任务难以标准化管理。业务方大多希望使用 SQL，避免使用 JAR 包。

我们调研了平台已有的 Datastream JAR 包任务，发现大部分的 JAR 包任务还是以 Table API 为主，只有少量过程用 Datastream 做了一些数据的转换，完成之后还是注册成了 Table 进行 Table 操作。如果平台可以支持在 SQL 里面做一些复杂的自定义转换，业务其实完全不需要编写代码。

因此我们支持了一种新的任务构建模式——算子化，模块化地构建一个 Flink 任务，混合 JAR 包与 SQL，在进行任务构建时，先定义一段 SQL，再定义一个 JAR 包，再接一段 SQL，每段都称为算子，算子之间相互串联，构成一个完整的任务。

采用 Flink 标准的 SQL 语法，对 JAR 包进行了接口的限制，必须继承平台的接口定义进行开发。输入输出都是定义好的 Datastream。它比 UDF 的扩展性更强，灵活性也更好。而且整个任务的输入输出基本可以做到和 SQL 同级别的管控力，算子的开发也比纯 JAR 包简单得多，不需要学习太多 Flink API 的操作，只需要对 Datastream 进行变换。而且对于一些常用的公共算子，平台可以统一开发提供，拥有更专业的性能优化，业务方只要引用即可。

目前在实时数仓等一些偏固定业务的场景，我们都在尝试进行标准化算子的推广和使用。

1.3 智能诊断

平台建设的第三点是流任务的智能诊断。目前实时支持的业务场景包括 ETL、AI、数据集成等，且任务规模增长速度很快。越来越大的规模对平台的服务能力也提出了更高的要求。

此前，平台人员需要花费很多的时间在协助业务解决资源或各种业务问题上，主要存在以下几个方面的问题：

资源配置：初始资源确认困难，碎片化严重，使用资源周期性变化;性能调优：数据倾斜，网络资源优化，state 性能调优，gc 性能调优;错误诊断：任务失败原因分析，修复建议。

这些问题日常都靠平台人员兜底，规模小的时候大家勉强可以负担，但是规模快速变大后已经完全无力消化，需要一套自动化的系统来解决这些问题。

因此我们做了一套流任务的智能诊断系统，架构如上图。

系统会持续抓取任务运行时的 metrics 进行性能分析，分析完成后推给用户，让用户自己执行具体的优化改进操作;也会实时抓取任务失败的日志，并与词库进行匹配，将错误进行翻译，使用户更容易理解，同时也会给出更好理解的解决方案，让用户自行进行故障处理;同时还会根据任务的历史运行资源进行自动化缩容处理，解决资源浪费和资源不足的问题。

目前此功能已经节省了整个队列 10% 的资源左右，分担了相当一部分平台的运维压力，在未来我们会持续进行优化迭代，更进一步提高这套系统在自动化运维上面的能力以及覆盖度。

未来，在提交引擎方面，我们希望融合 Yarn session 模式与 application 模式做 session 的复用，解决任务上线的资源申请效率问题。同时希望大 state 任务也能够在 session 的基础上复用本地的 state，启动时无需重新下载 state。

智能诊断方面，我们希望实现更多自动化的操作，实现自动进行优化改进，而不需要用户手动操作，做到用户低感知;扩容缩容也会持续提速，目前缩容的频率只在天级，扩容还未实现自动化。未来我们希望整个操作的周期和频率做到分钟级的自动化。

算子方面，我们希望能统一目前的 SQL 和 JAR 包两种模式，统一任务构建方式，让用户以更低的成本更多复杂的操作，平台也更方便管理。

二、增量化

上图是我们早期的数据架构，是典型的 Lambda 架构。实时和离线从源头上就完全分离、互不干涉，实时占较低，离线数仓是核心的数仓模型，占主要的比例，但它存在几个明显的问题。

第一，时效性。数仓模型是分层架构，层与层之间的转换靠调度系统驱动，而调度系统是有周期的，常见的基本都是天或小时。源头生产的数据，数仓各层基本需要隔一天或几个小时才可见，无法满足实时性要求稍高的场景;

第二，数据的使用效率低。ETL 和 adhoc 的数据使用完全一样，没有针对性的读写优化，也没有按照用户的查询习惯进行重新组织，缺乏数据布局优化的能力。

针对第一个问题，是否全部实时化即可?但是实时数仓的成本高，而且不太好做大规模的数据回溯。大部分业务也不需要做到 Kafka 的秒级时效。第二个问题也不好解决，流式写入为了追求效率，对数据的布局能力较弱，不具备数据的重新组织能力。因此我们在实时和离线之间找到了一个平衡——做分钟级的增量化。

我们采用 Flink 作为计算引擎，它的 checkpoint 是一个天然的增量化机制，实时任务进行一次 checkpoint，产出一批增量数据进行增量化处理。数仓来源主要有日志数据和 binlog 数据，日志数据使用 Append 传统的 HDFS 存储即可做到增量化的生产;binlog 数据是 update 模式，但 HDFS 对 update 的支持并不好，因此我们引入了 Hudi 存储，它能够支持 update 操作，并且具备一定的数据布局能力，同时它也可以做 Append 存储，并且能够解决 HDFS 的一些小文件问题。因此日志数据也选择了 Hudi 存储，采用 Append 模式。

最终我们的增量化方案由 Flink 计算引擎 + Hudi 存储引擎构成。

增量化场景的落地上，考虑到落地的复杂性，我们先选取了业务逻辑相对简单、没有复杂聚合逻辑的 ODS 和 DWD 层进行落地。目前的数据是由 Flink 直接写到 Hive 的 ODS 层，我们对此进行了针对性的适配，支持了 Hive 表的增量化读取，开发了 HDFSStreamingSource，同时为了避免对 HDFS 路径频繁扫描的压力，ODS 层写入时会进行索引创建，记录写入的文件路径和时间，只需要追踪索引文件即可。

source 也是分层架构，有文件分发层和读取层，文件分发层进行协调，分配读取文件数，防止读取层某个文件读取过慢堆积过多文件，中间的转换能够支持 FlinkSQL 操作，具备完整的实时数仓的能力。

sink 侧我们引入了 Hudi connector，支持数据 Append 写入 Hudi，我们还对 Hudi 的 compaction 机制进行了一些扩展，主要有三个：DQC 检测、数据布局的优化以及映射到 Hive 表的分区目录。目前数据的布局依旧还很弱，主要依赖 Hudi 本身的 min、max 和 bloom 的优化。

完成所有上述操作后，ODS 到 DWD 的数据时效性有了明显提升。

从数据生产到 DWD 可见，提高到了分钟级别;DWD 层的生产完成时间也从传统的 2:00~5:00 提前到了凌晨 1 点之前。此外，采用 Hudi 存储也为日后的湖仓一体打下了以一个好的基础。

除了日志数据，我们对 CDC 也采用这套方案进行加速。基于 Flink 的 CDC 能力，针对 MySQL 的数据同步实现了全增量一体化操作。依赖 Hudi 的 update 能力，单任务完成了 MySQL 的数据同步工作，并且数据只延迟了一个 checkpoint 周期。CDC 暂时不支持全量拉取，需要额外进行一次全量的初始化操作，其他的流程则完全一致。

Hudi 本身的模型和离线的分区全量有较大的区别，为了兼容离线调度需要的分区全量数据，我们也修改了 Hudi 的 compaction 机制。在做划分区的 compaction 时会做一次数据的全量拷贝，生成全量的历史数据分区，映射到 Hive 表的对应分区。同时对于 CDC 场景下的数据质量，我们也做了很多的保障工作。

为了保证 CDC 数据的一致性，我们从以下 4 个方面进行了完善和优化：

第一，binlog 条数的一致性。按照时间窗口进行 binlog 生产侧和消费侧的条数校验，避免中间件丢数据;

第二，数据内容抽样检测。考虑到成本，我们在 DB 端和源端、Hudi 存储端抽样增量数据进行内容的精确比较，避免 update 出错;

第三，全链路的黑盒测试。测试库表模拟了线上情况，进行 7×24 小时不间断的 Kafka 生产 MySQL 数据，然后串通整套流程防止链路故障;

第四，定期的全量对比。业务的库表一般比较大，历史数据会低频地定期进行全量比对，防止抽样观测漏掉的错误。

刚开始使用 Hudi 的时候，Hudi on Flink 还是处于初级的阶段，因此存在大量问题，我们也一起和 Hudi 社区做了大量优化工作，主要有 4 个方面：Hudi 表的冷启动优化、checkpoint 一致性问题解决、Append 效率低的优化以及 get list 的性能问题。

首先是冷启动的问题。Hudi 的索引存储在 Flink state 里，一张存在的 Hudi 表如果要通过 Flink 进行增量化更新写入，就必然面临一个问题：如何把 Hudi 表已有的信息写入到 Flink state 里。

MySQL 可以借助 Flink CDC 完成全量 + 增量的过程构建，可以绕开从已有 Hudi 表冷启动的过程，但是 TiDB 不行，它的存量表在借助别的手段构建完之后，想要增量化就会面临如何从 FlinkSQL 冷启动的问题。

社区有个原始方案，在记录所有的算子 BucketAssigner 里面读取全部的 Hudi 表数据，然后进行 state 构建，从功能上是可行的，但是在性能上根本无法接受，尤其是大表，由于 Flink 的 key state 机制原理，BucketAssigner 每个并发度都要读取全表数据，然后挑选出属于当前这个并发的数据存储到自己的 state 里面，每个并方案都要去读全量的表，这在性能上难以满足。

业务能启动的时间太长了，很多百亿级别的表能启动的时间可能是在几个小时，而且读取的数据太多，很容易失败。

和社区进行了沟通交流后，他们提供了一套全新的方案，新增了独立的 Bootstrap 机制，专门负责冷启动过程。Bootstrap 由 coordinator 和 IndexBootstrap 两个算子组成，IndexBootstrap 负责读取工作，coordinator 负责协调分配文件读取，防止单个 IndexBootstrap 读取速度慢而降低整个初始化流程的效率。

IndexBootstrap 算子读取到数据后，会按照与业务数据一样的 Keyby 规则，Keyby 到对应的 BucketAssigner 算子上，并在数据上面打标，告知 BucketAssigner 这条数据是有 Bootstrap 的，不需要往下游 writer 发送。整个流程里，原始数据只需读取一遍，而且是多并发一起读，效率获得了极大的提升。而且 BucketAssigner 只需要处理自己应该处理的数据，不再需要处理全表的数据。

其次是 Hudi 的 checkpoint 一致性问题。Hudi on checkpoint 在每次 checkpoint 完成的时候会进行一次 commit 操作，具体流程是 writer 算子在 checkpoint 的时候 flush 内存数据，然后给 writer coordinator 算子汇报汇总信息，writer coordinor 算子收到汇报信息时会将其缓存起来，checkpoin 完成后，收到 notification 信息时会进行一次 commit 操作。

但是在 Flink 的 checkpoint 机制里，notification 无法保证一定成功，因为它并不在 checkpoint 的生命周期里，而是一个回调操作，是在 checkpoin 成功后执行。checkpoin 成功后，如果这个接口还没有执行完成，commit 操作就会丢失，也就意味着 checkpoint 周期内的数据会丢失。

针对上述问题，我们进行了重构。Writer 算子在 cehckpoint 时，会对汇报的 writer coordinator 的信息进行 state 持久化，任务重启后重新汇报给 writer coordinator 算子。writer coordinator 算子再收集所有 writer 算子信息并做一次 commit 判断，确保对应的 commit 已经完成。此时，Writer 算子也会保持阻塞，确保上次持久化的 commit 完成之后才会处理最新的数据，这样就对齐了 Hudi 与 Flink 的 checkpoint 机制，保证了边界场景数据的一致性。

第三是针对 Hudi 在 Append 写入场景下的优化。

由于 Append 模式是复用 update 模式的代码，所以在没有重复 key 的 Append 场景下，很多操作是可以简化的，因为 update 为了处理重复，需要做很多额外的操作。如果能够简化这些操作，吞吐能力可以有较大的提升。

第一个操作是小文件的查找，每次 checkpoint 后，update 都会重新 list 文件，然后从文件中找到大小不达标的文件继续 open 并写入。update 场景存在倾斜，会造成很多文件大小不均匀，但是 Append 场景不存在这种问题，它所有的文件大小都很均匀;

第二个是 keyby。在 update 的模式下面，单个 key 只能被一个节点处理，因此上游需要按照 Hudi key 进行 keyby 操作。但是 Append 场景没有重复 key，可以直接用 chain 代替 keyby，大大减少了节点之间序列化传输的开销。同时 Append 场景下不存在内存合并，整体效率也会更高。

最后一个是 GetListing 的优化。Hudi 表与底层 HDFS 文件的映射是通过 ViewManager 来做的，Hudi table 对象和 TimelineService 都会自己去初始化一个 ViewManager，每个 ViewManager 在初始化的时候都会进行 HDFS 目录的 list 操作，由于每个并发都持有多个 Hudi table 或 TimelineService，会造成大并发任务启动时 HDFS 的压力很大。我们对 TimelineService 进行了单例化的优化，保证每个进程只有一 TimelineService，能够数倍地降低 HDFS list 的压力。后续我们还会基于 Flink 的 coordinator 机制做任务级别的单例化。

未来，我们会继续挖掘增量的能力，给业务带来更多的价值。

三、AI on Flink

传统的机器学习链路里数据的传输、特征的计算以及模型的训练，都是离线处理的，存在两个大的问题。

第一个是时效性低，模型和特征的更新周期基本是 t+1 天或者 t+1 小时，在追求时效性的场景下体验并不好。第二个是计算训练的效率很低，必须等天或小时的分区数据全部准备好之后才能开始特征计算和训练。全量分区数据导致计算和训练的压力大。

在实时技术成熟后，大部分模型训练流程都切换到实时架构上，数据传输、特征计算和训练都可以做到几乎实时，从全量变成了短时的小批量增量进行，训练的压力也大大减轻。同时由于实时对离线的兼容性，在很多场景比如特征回补上，也可以尝试使用 Flink 的流批一体进行落地。

上图是我们典型的机器学习链路图。从图上可以看出，样本数据生产特征的计算、模型的训练和效果的评估都大量实时化，中间也夹杂着少量离线过程，比如一些超长周期的特征计算。

同时也可以看出，完整的业务的模型训练链路长，需要管理和维护大量的实时任务和离线任务。出现故障的时候，具体问题的定位也异常艰难。如何在整个机器学习的链路中同时管理号这么多实时和离线任务，并且让任务之间的协同和调度有序进行、高效运维，是我们一直在思考的问题。

因此我们引入了 Flink 生态下 AIFlow 系统。AIFlow 本身的定位就是做机器学习链路的管理，核心的机器计算引擎是 Flink，这和我们的诉求不谋而合。这套系统有三个主要的特性符合我们的业务需求。

第一，流批的混合调度。在我们实际的业务生产上，一套完整的实时链路都会夹杂着实时和离线两种类型的任务。AIFlow 支持流批的混合调度，支持数据依赖与控制依赖，能够很好地支持我们现有的业务形态。并且未来在 Flink 流批一体方面也会有更多的发挥空间;

第二，元数据的管理，AIFlow 对所有数据和模型都支持版本管理。有了版本管理，各种实验效果和实验参数就都可追溯;

第三，开放的 notification 机制。整个链路中存在很多的外部系统节点，难以归纳到平台内部，但是通过 notification 机制，可以打通 AIFlow 内部节点与外部节点的依赖。整套系统的部署分为三部分，notification service、 meta service 以及 scheduler，扩展性也很好，我们在内部化的过程中实现了很多自己的扩展。

实时平台在今年引入 AIFlow 的之后已经经历了两个版本的迭代，V2 版本是社区 release 之前的一个内部版本，我们进行了分装提供试用。V3 版本是今年 7 月社区正式 release 之后，我们进行了版本的对接。

AIFlow 的构建使用 Python 进行描述，运行时会有可视化的节点展示，可以很方便地追踪各个节点的状态，运维也可以做到节点级的管理，不需要做整个链路级别的运维。

未来我们会对这套系统在流批一体、特征管理以及模型训练三个方向进行重点的迭代与开发，更好地发挥它的价值。

关键词： 机器学习 任务失败 输入输出 可以看出