详解 Flink 实时应用的确定性-阿里云开发者社区

作者：林小铂（网易游戏）

确定性（Determinism）是计算机科学中十分重要的特性，确定性的算法保证对于给定相同的输入总是产生相同的输出。在分布式实时计算领域，确定性是业界一直难以解决的课题，由此导致用离线计算修正实时计算结果的 Lambda 架构成为大数据领域过去近十年的主流架构。

而在最近几年随着 Google The Dataflow Model 的提出，实时计算和离线计算的关系逐渐清晰，在实时计算中提供与离线计算一致的确定性成为可能。本文将基于流行实时计算引擎 Apache Flink，梳理构建一个确定性的实时应用要满足什么条件。

确定性与准确性

比起确定性，准确性（Accuracy）可能是我们接触更多的近义词，大多数场景下两者可以混用，但其实它们稍有不同: 准确的东西一定是确定的，但确定性的东西未必百分百准确。在大数据领域，不少算法可以根据需求调整成本和准确性的平衡，比如 HyperLogLog 去重统计算法给出的结果是有一定误差的（因此不是准确的），但却同时是确定性的（重算可以得到相同结果）。

要分区确定性和准确性的缘故是，准确性与具体的业务逻辑紧密耦合难以评估，而确定性则是通用的需求（除去少数场景用户故意使用非确定性的算法）。当一个 Flink 实时应用提供确定性，意味着它在异常场景的自动重试或者手动重流数据的情况下，都能像离线作业一般产出相同的结果，这将很大程度上提高用户的信任度。

影响 Flink 应用确定性的因素

投递语义

常见的投递语义有 At-Most-Once、At-Least-Once 和 Exactly-Once 三种。严格来说只有 Exactly-Once满足确定性的要求，但如果整个业务逻辑是幂等的，基于At-Least-Once 也可以达到结果的确定性。

实时计算的 Exactly-Once 通常指端到端的 Exactly-Once，保证输出到下游系统的数据和上游的数据是一致的，没有重复计算或者数据丢失。要达到这点，需要分别实现读取数据源（Source 端）的 Exactly-Once、计算的 Exactly-Once 和输出到下游系统（Sink 端）的 Exactly-Once。

其中前面两个都比较好保证，因为 Flink 应用出现异常会自动恢复至最近一个成功 checkpoint，Pull-Based 的 Source 的状态和 Flink 内部计算的状态都会自动回滚到快照时间点，而问题在于 Push-Based 的 Sink 端。Sink 端是否能顺利回滚依赖于外部系统的特性，通常来说需要外部系统支持事务，然而不少大数据组件对事务的支持并不是很好，即使是实时计算最常用的 Kafka 也直到 2017 年的 0.11 版本才支持事务，更多的组件需要依赖各种 trick 来达到某种场景下的 Exactly-Once。

总体来说这些 trick 可以分为两大类:

依赖写操作的幂等性。比如 HBase 等 KV 存储虽然没有提供跨行事务，但可以通过幂等写操作配合基于主键的 Upsert 操作达到 Exactly-Once。不过由于 Upsert 不能表达 Delete 操作，这种模式不适合有 Delete 的业务场景。
预写日志（WAL，Write-Ahead-Log）。预写日志是广泛应用于事物机制的技术，包括 MySQL、PostgreSQL 等成熟关系型数据库的事物都基于预写日志。预写日志的基本原理先将变更写入缓存区，等事务提交的时候再一次全部应用。比如 HDFS/S3 等文件系统本身并不提供事务，因此实现预写日志的重担落到了它们的用户（比如 Flink）身上。通过先写临时的文件/对象，等 Flink Checkpoint 成功后再提交，Flink 的 FileSystem Connector 实现了 Exactly-Once。然而，预写日志只能保证事务的原子性和持久性，不能保证一致性和隔离性。为此 FileSystem Connector 通过将预写日志设为隐藏文件的方式提供了隔离性，至于一致性（比如临时文件的清理）则无法保证。

为了保证 Flink 应用的确定性，在选用官方 Connector，特别是 Sink Connector 时，用户应该留意官方文档关于 Connector 投递语义的说明[3]。此外，在实现定制化的 Sink Connector 时也需要明确达到何种投递语义，可以参考利用外部系统的事务、写操作的幂等性或预写日志三种方式达到 Exactly-Once 语义。

函数副作用

函数副作用是指用户函数对外界造成了计算框架意料之外的影响。比如典型的是在一个 Map 函数里将中间结果写到数据库，如果 Flink 作业异常自动重启，那么数据可能被写两遍，导致不确定性。对于这种情况，Flink 提供了基于 Checkpoint 的两阶段提交的钩子（CheckpointedFunction 和 CheckpointListener），用户可以用它来实现事务，以消除副作用的不确定性。另外还有一种常见的情况是，用户使用本地文件来保存临时数据，这些数据在 Task 重新调度的时候很可能丢失。其他的场景或许还有很多，总而言之，如果需要在用户函数里改变外部系统的状态，请确保 Flink 对这些操作是知情的（比如用 State API 记录状态，设置 Checkpoint 钩子）。

Processing Time

在算法中引入当前时间作为参数是常见的操作，但在实时计算中引入当前系统时间，即 Processing Time，是造成不确定性的最常见也最难避免的原因。对 Processing 的引用可以是很明显、有完善文档标注的，比如 Flink 的 Time Characteristic，但也可能是完全出乎用户意料的，比如来源于缓存等常用的技术。为此，笔者总结了几类常见的 Processing Time 引用:

Flink 提供的 Time Characteristic。Time Characteristic 会影响所有使用与时间相关的算子，比如 Processing Time 会让窗口聚合使用当前系统时间来分配窗口和触发计算，造成不确定性。另外，Processing Timer 也有类似的影响。
直接在函数里访问外部存储。因为这种访问是基于外部存储某个 Processing Time 时间点的状态，这个状态很可能在下次访问时就发生了变化，导致不确定性。要获得确定性的结果，比起简单查询外部存储的某个时间点的状态，我们应该获取它状态变更的历史，然后根据当前 Event Time 去查询对应的状态。这也是 Flink SQL 中 Temporary Table Join 的实现原理[1]。
对外部数据的缓存。在计算流量很大的数据时，很多情况下用户会选择用缓存来减轻外部存储的负载，但这可能会造成查询结果的不一致，而且这种不一致是不确定的。无论是使用超时阈值、LRU（Least Recently Used）等直接和系统时间相关的缓存剔除策略，还是 FIFO（First In First Out）、LFU（Less Frequently Used）等没有直接关联时间的剔除策略，访问缓存得到的结果通常和消息的到达顺序相关，而在上游经过 shuffle 的算子里面这是难以保证的（没有 shuffle 的 Embarrassingly Parallel 作业是例外）。
Flink 的 StateTTL。StateTTL 是 Flink 内置的根据时间自动清理 State 的机制，而这里的时间目前只提供 Processing Time，无论 Flink 本身使用的是 Processing Time 还是 Event Time 作为 Time Characteristic。BTW，StateTTL 对 Event Time 的支持可以关注 FLINK-12005[2]。

综合来讲，要完全避免 Processing Time 造成的影响是非常困难的，不过轻微的不确定性对于业务来说通常是可以接受的，我们要做的更多是提前预料到可能的影响，保证不确定性在可控范围内。

Watermark

Watermark 作为计算 Event Time 的机制，其中一个很重要的用途是决定实时计算何时要输出计算结果，类似文件结束标志符（EOF）在离线批计算中达到的效果。然而，在输出结果之后可能还会有迟到的数据到达，这称为窗口完整性问题（Window Completeness）。

窗口完整性问题无法避免，应对办法是要么更新计算结果，要么丢弃这部分数据。因为离线场景延迟容忍度较大，离线作业可以推迟一定时间开始，尽可能地将延迟数据纳入计算。而实时场景对延迟有比较高的要求，因此一般是输出结果后让状态保存一段时间，在这段时间内根据迟到数据持续更新结果（即 Allowed Lateness），此后将数据丢弃。因为定位，实时计算天然可能出现更多被丢弃的迟到数据，这将和 Watermark 的生成算法紧密相关。

虽然 Watermark 的生成是流式的，但 Watermark 的下发是断点式的。Flink 的 Watermark 下发策略有 Periodic 和 Punctuated 两种，前者基于 Processing Time 定时触发，后者根据数据流中的特殊消息触发。

图1. Periodic Watermark 正常状态与重放追数据状态

基于 Processing Time 的 Periodic Watermark 具有不确定。在平时流量平稳的时候 Watermark 的提升可能是阶梯式的（见图1(a)）；然而在重放历史数据的情况下，相同长度的系统时间内处理的数据量可能会大很多（见图1(b))，并且伴有 Event Time 倾斜（即有的 Source 的 Event Time 明显比其他要快或慢，导致取最小值的总体 Watermark 被慢 Watermark 拖慢），导致本来丢弃的迟到数据，现在变为 Allowed Lateness 之内的数据（见图1中红色元素）。

图2. Punctuated Watermark 正常状态与重放追数据状态

相比之下 Punctuated Watermark 更为稳定，无论在正常情况（见图2(a)）还是在重放数据的情况（见图2(b)）下，下发的 Watermark 都是一致的，不过依然有 Event Time 倾斜的风险。对于这点，Flink 社区起草了 FLIP-27 来处理[4]。基本原理是 Source 节点会选择性地消费或阻塞某个 partition/shard，让总体的 Event Time 保持接近。

除了 Watermark 的下发有不确定之外，还有个问题是现在 Watermark 并没有被纳入 Checkpoint 快照中。这意味着在作业从 Checkpoint 恢复之后，Watermark 会重新开始算，导致 Watermark 的不确定。这个问题在 FLINK-5601[5] 有记录，但目前只体现了 Window 算子的 Watermark，而在 StateTTL 支持 Event Time 后，或许每个算子都要记录自己的 Watermark。

综上所述，Watermark 目前是很难做到非常确定的，但因为 Watermark 的不确定性是通过丢弃迟到数据导致计算结果的不确定性的，只要没有丢弃迟到数据，无论中间 Watermark 的变化如何，最终的结果都是相同的。