摘要： 本文由快手大数据架构团队负责人赵健博分享，主要介绍 Apache Flink 在快手的过去、现在和未来。内容包括：

  1. 为什么选 Flink
  2. Flink 在快手的发展
  3. 业务数据流
  4. 技术创新
  5. 未来计划

大家好，我是赵健博，来自快手，目前负责快手大数据架构团队。今天很高兴可以和大家分享我们在 Flink 项目上的应用、改进与发展历程。

先来看一下我们选择 Flink 引擎的主要原因：

  • 首先，Flink 能做到亚秒级处理延迟。目前大部分的业务需求对实时处理延迟要求越来越高，这是个最基本需求。
  • 其次，Flink 有丰富的窗口计算模式，且自带状态存储引擎以及精准一次的语义，这个能力极大简化了数据的处理复杂度，显著提升了研发的速度。
  • 最后，批流一体能力以及研发模式的变革，也将进一步提效研发，为业务赋能。

本次会议也看到了很多公司都在分享批流一体落地实践，相信流批一体全场景落地的大进程也将指日可待。

Flink 在快手的发展历程，总的来说可以分为四个阶段：

  1. 我们是从 17 年开始使用 Flink 的，17 年我们主要是初步试用，当时接入的业务是直播与短视频的质量监控业务。
  2. 进入到 2018 年之后，在能力上，我们开始对 Flink 进行成周边体系的建设，例如，构建引擎内部 metric 的采集，监控与报警流程、作业托管平台上线等。与此同时，我们也在不断的加深对 Flink 的理解，修炼内功；在业务上，开始接入直播 CDN 流量调度，日志实时拆分、投放分析、客户端 Crash 分析等场景。
  3. 进入到 2019 年后，随着对 Flink 引擎掌控力的加强，我们开始进行一些稳定性与性能相关的改进，主要包括防雪崩，流控、分级保障、参数热更新、自研状态存储引擎 Slimbase、实时多维建模等。在业务上，开始支撑春节活动大屏、实时多维分析、曝光/点击流实时 Join 等场景。
  4. 到 2020 年后，我们除了持续关注稳定性性能之外，也在推进效率改进，例如调研并开始试用 Flink SQL，以及流批一体能力。在业务上，采用 Flink SQL 支撑活动大屏、开始通过 Flink 以及流批一体能力建设 AI 数据流、实时报表、直播精彩时刻等业务场景。

截止到目前，快手 Flink 从业务规模上看有若干集群，集群有数千机器，目前还是部署在 YARN 上，后续也会考虑迁移到 K8s 上。总的作业 2000 左右，这些作业每天处理 20 多万亿条的记录，其中峰值达到每秒 6 亿条的规模。

三、业务数据流

1. 数据流的总体架构图

接下来，让我们看下快手 Flink 目前应用的一些业务场景与业务数据流的案例。

下面这张图是一个数据流的总体架构图，从这张图中，大家能看到数据的源头有三类数据，一个是数据库中的数据，一个是服务端的日志，最后是客户端的日志，这些日志上报给 Kafka 的服务。

在快手，所有日志或者消息都是通过 Kafka 服务流转的。数据进入到 Kafka 之后分别流转到实时数据链路，以及离线数据链路上（实时同步到 Hive）。在实时链路上，目前 Flink 支撑了很多业务场景，如：实时 ETL、数据集成、实时报表计算、实时监控、实时实时特征等等。这些数据通过 Flink 实时计算处理之后，将流入到各种类型的数据库中，例如多维数据库（Druid/Clickhouse），MySQL、Redis、HBase 等等。之后各类的数据产品、数据应用、业务应用从这些数据库中获取最新的聚合或者结果数据，进行业务的处理。

2. 实时 ETL 场景

接下来，我们展开介绍下上述各个场景下的业务数据流图。在实时 ETL 场景下，目前我们主要在推广使用 Flink SQL 进行数据的实时 ETL。下图左侧展示了实时 ETL的流程，其中 Kafka 中的 topic 的 schema 都被元数据服务管理起来了。Flink 引擎首先访问元数据中心获取 Topic 的 schema，然后将 Topic 转成实时表，并通过 SQL 完成 ETL 的处理落地。右侧的 SQL 是我们进行数据拆分的案例。

3. 数据集成场景

在数据集成场景下，如左图所示，通过 Flink 引擎可以很方便地完成 Kafka/HBase/ES/Hive/Redis 等服务的数据交换。相比于其他引擎，Flink 的 source/sink 支持的服务种类更丰富，且更加方便扩展。除此之外，除了离线数据交换，Flink 是天然支撑实时场景的。

4. 实时报表的场景

在实时报表的场景下，介绍下 Flink 支持快手春节活动的实时数据链路。

如图所示，整个数据流从左到右共分为 4 层，分别是 ODS 层、采样层、指标逻辑计算层、数据服务层。

  1. 最开始是原始的 ODS 层数据，通过客户端，服务端，或者是 DB 直接打到 Kafka 的 topic 中形成一个 ODS 层，这一层的数据经过 Flink 的处理，再写回 Kafka，形成一个采样层。
  2. 采样层提出来的原因主要是，面向春节活动的流量高峰，没法精准预知它的峰值有多高，所以我们需要具备对整个流量进行采样的能力，以便能够在有限的资源下应对洪峰。一旦洪峰来了，可以进行数据采样处理，有效降低计算资源的消耗，同时再通过采样的规则在后续逻辑计算层还原采样之前数据指标的结果。
  3. 数据被采样之后再通过 Flink 进行逻辑层的计算，例如留存、新增、PV、UV 等指标，然后将这些指标最终保存到 Redis 或者多维引擎中。在这个计算过程中，当时采用的是外部存储与服务进行了 UV，以及新增的计算。在未来的活动支撑中，我们会逐渐替换为 Flink 自己的 state 引擎。
  4. 最后，各类数据产品与服务，如大屏，看板等，从 Redis 或者多维引擎中获取数据进行展示以及策略的调整。

5. 实时监控场景

在实时监控这个场景下，介绍下快手直播质量监控和 CDN 流量调度链路。

首先数据通过埋点采集，打到 Kafka 之后，在实时链路的处理上，通过 Flink 进行数据的清洗、转换、聚合，形成 DWD 和 DWS 层的数据，这些数据也会最终写回 Kafka。之后，会把 DWS 层的数据导到后面的 OLAP 这种数据库中。然后上层的 BI 服务通过访问数据库中的数据进行报表的展示，从而完成监控，以及数据决策。同时，质量计算的调度结果数据存储 Redis，供在线 CDN 调度服务提供决策依据。

在离线链路上，可以考虑从实时链路中的每一层进行数据的导出，导到 Hive 表中。这部分数据的保存主要是为了解决 Ad hoc 分析，以及当实时流数据出现问题，进行的离线的数据修正。

6. 特征处理场景

最后一个业务场景，介绍一个 AI 数据流案例，特征处理与索引生成流程。在快手，有大量的特征需要处理，特征的处理与管理效率对模型迭代效率有很大影响。采用 Flink 进行特征与索引的处理，在管理上与研发效率上都有比较大的优势。

目前我们借助 Flink 完成了一部分的特征与索引生产流程，如图所示，行为数据通过 Kafka 流入 Flink 之后，利用 Flink 的窗口计算能力完成各种类型的特征实时计算，之后将特征存储到特征库中，同时也会同步一份数据到 Hive 中，用作做特征离线数据流处理；除此之外，当有索引需要生成的时候，会通过 Kafka 触发生成策略，下游的索引生成的 Flink 作业从各种特征库中获取特征并进行处理后，形成索引，存储到索引库中。最终的索引数据，为在线的推荐服务提供召回源。

四、技术创新

1. 状态引擎

接下来重点介绍一下 Flink 在快手做的一些技术改进和创新。首先介绍下我们自研的状态引擎 Slimbase。它在设计上分了三层：

  • 接口层，在接口层主要兼容目前状态存储的几类接口，value、list、map 状态等。
  • 中间层，我们构建了一个 KV 的 cache 层，主要是做数据的读和写的加速。在这层内部，又分为高速 KV 层和 Chunk 层，高速 KV 层（HashMap）有非常快的存取速度，但是空间利用率比较低。为了节省空间，我们又在整个高速 KV 层下面建了一个 Chunk 层，一个 Chunk 是多个 KV 序列化组成的。通过这种序列化的组织之后，在某些场景下相比于 KV 层能够节省约 60% 的空间。但是在存取速度上会有一定程度的降低。实际使用的时候，可以根据实际情况灵活控制高速 KV 层与 Chunk 层的容量配比。
  • 分布式文件系统层，缓存层被淘汰的数据将会写入到文件系统层，最终形成一个个文件。为了提高文件系统层面的读取性能，多个文件会通过 compaction 进行合并。此外，文件系统层有文件块级别的缓存，具备缓存热点数据能力

以上就是 Slimbase 整体架构。下面我们看看 Flink Benchmark 跑出来结果（和RocksDB 对比）。本次测试采用了相同大小的缓存，数据集采用了50w、1500w、5000w 三种规模。

目标是测试三种场景下的结果：

  • 仅覆盖高速 KV 缓存；
  • 覆盖高速 KV 缓存 +Chunk 缓存；
  • 覆盖 KV 高速缓存 +Chunk 缓存+文件系统；

这是 50 万的数据集，这些数据集全部是在高速的 KV 层中的。从测试结果上看，相比 RocksDB，Slimbase 读写有 3~9 倍的性能提升。

在 1500w 数据规模下，数据会分布在高速的 KV 层加 Chunk 层，相比 RocksDB，读写有 2~6 倍的性能提升。

在 5000w 数据规模下，数据命中的层次变得更多，把文件系统也覆盖到了。相比前两个场景，我们发现性能有比较大的下降。相比 RocksDB，读性能 0.50.7；写性能 0.904 倍。所以我们接下来会在整个文件系统层的存取性能上，会做专项的优化，提升整个文件系统的性能，最终可以超过 RocksDB 性能。

2. 稳定性

在介绍稳定性的改进前，我们先来看一下影响 Flink 稳定性的因素有哪些。我这里总结了三点：

  • 硬件故障，例如机器故障，机柜故障，Tor 故障，机房故障等。
  • Flink 依赖的服务异常，例如 Kafka 集群异常，HDFS 服务异常等。
  • Flink 流量过载，例如硬件满载，以及由于数据源消费速度差异导致的满载。

在硬件故障场景下，这里面取了一个单点的场景。看下这个 Flink 作业，由两个 source，一个 window 组成。右侧是 Flink 作业的物理部署的情况。最大的框代表一台机器，大框里面的多个小框代表多个 TaskManager。

如果出现了一个节点故障，比如 node3 发生故障了。Flink 引擎会重新从 YARN 申请资源，完成 TaskManager 初始化，并重新部署作业。

我们对一个业务作业做了一个分析，发现宕机故障后到作业恢复，共需要 90s 的时间。宕机检测 (60秒)，重新申请资源容器 (5秒)，容器初始化 (20秒)，作业重新部署执行 (5秒）。这对于某些在线业务场景来说是不能接受的。从具体的过程拆解来看，发现宕机检测和初始化的消耗是大头。要如何改进呢？

从解决思路上来说，包含两个方面。首先 60 秒的宕机检测，时间太长了。对此，要做到快速发现宕机。此外，还要预留资源，当宕机出现时，可以省去申请资源，以及初始化的时间。

在宕机快速发现方面，我们研发了 Hawk Service，它是一个多数派的连通性检测服务，具体的检测流程是 Hawk 集群中多个工作节点会周期性地检测集群中每台机器的连通性，由于它是多数派的，所以可信度是有保障的。最终，Hawk 服务可以做到在 10 秒钟之内发现一个宕机事件。

此外，在预留资源方面，我们扩展了 Flink 作业的资源申请模型，在 Flink 提交时可以设定一个资源冗余参数，当冗余参数被激活后，会自动保障冗余资源量会高于单点故障导致的资源缺失量，且在资源排布上避免冗余资源的聚集性。如图所示：

有了这两点能力之后，如果同样是第三台机器挂掉了，我们能在 10 秒内发现。并且由于资源已经分配好了，直接部署一遍作业就可以了。所以整体的恢复时间从 4 个步骤直接缩短为 2 个步骤，时间上从 90s 可以缩短到 15s 左右。

接下来，我们看看如果 Flink 引擎依赖的服务异常了要怎么办呢？这里举了一个 Kafka 服务异常的例子。还是同样的 Flink 的作业，依赖两个 topic，Flink 作业在 B 机房，读取的 Kafka 也在 B 机房，写入的 Kafka 在 A 机房。如果出现读取或者写入的 Kafka 集群异常了，Flink 作业需要具备 Failover Kafka 集群的能力，当然如果是切读，Kafka 的上游也需要联动切流。

在过载场景下，我举了两个例子：

  • 不同数据源快慢消费导致满载

在这个 case 中，消费 topicA 的 source 速度慢，消费 topicB 的数据源快，由于后边存在 window 操作，会导致 window 的状态持续变大，最终引导作业不稳定。这个问题要如何解决呢？

我们采用的办法是同步所有相关数据源消费的进度，引入一个 source 的协调者（SourceCoordinator），周期性收集 source 源 waterwark 的进展，并根据全局的现状，预测出来各个 source 源接下来允许读到的最大位置 target Watermark，之后下发给所有的 source，source 根据得到的 target Watermark 以及当前自己 watermark，确定读取速度。最终全局 source 达到同步读的结果，最小 source 和最大 source 的差距在一个可控制的范围内。

  • 硬件资源满载

如果硬件出现了满载要怎么处理呢？例如，其中一个 TM 所在的机器出现 CPU 满载了，或者大范围出现机器满载。

解决方案跟上面的是类似的，控制数据源的消费速度。如图所示，引入 HealthyCoordinator，周期性检查 TM 上的资源消耗情况，并根据负载限制 source 的消费速度。动态