深入理解流计算中的 Watermark _流计算

近年来流计算技术发展迅猛，甚至有后来居上一统原本批处理主导的分布式计算之势，其中 Watermark 机制作为流计算结果准确性和延迟的桥梁扮演着不可或缺的角色。然而由于缺乏高质量的学习资源加上计算 Watermark 确实不是一件容易的事情，不少有着批处理计算背景的用户在流计算作业的开发中可能并不理解 Watermark 的重要意义，从而多走了很多弯路。为此，本文将基于笔者的学习积累和开发经验，谈谈个人对 Watermark 的理解，希望起到抛砖引玉的作用。
本文将首先说明 Watermark 提出的背景，然后详细解析 Watermark 的原理，最后结合工业案例说明 Watermark 在实践中如何被应用。
Watermark 背景自 google 的三篇论文和 Hadoop 出现后，工业界的分布式计算技术进入了百花齐放的时期，然而相比于离线批处理计算的蓬勃发展，作为后来者的流计算却有点停滞不前。流计算和批处理在对于每条记录的单独处理上基本一致，不同之处在于聚合类的计算。批处理计算结果的输出依赖于输入数据集合的结束，而流计算的输入数据集通常是无边界的，不可能等待输入结束再输出结果。针对这个问题流处理引入了窗口的特性，简单来说就是将无限的数据流按照时间范围切分为一个个有限的数据集，所以我们依然能够沿用批处理的计算模型。来到这时，业界在流计算和批处理的关系上出现了两种截然不同的观点，一个观点认为流计算是批处理的特例，另一个观点则认为批处理是流处理的特例。
实时计算与离线计算的分离流计算是批处理的特例的观点在早期占据了主导的地位，其中最为典型的便是以 Spark Streaming 为代表的 micro-batching 类型的实时处理框架的流行。Micro-batching 的主要思想是以分钟甚至秒级别的执行间隔来将批处理应用到数据流上，但不久后人们意识到这种计算模型依然不能完全满足低延迟高准确性的要求，主要问题除了批处理调度导致的延迟外，还有一点是窗口变小后，数据收集延迟对结果准确性的影响大大增强了。比如说计算一个游戏服务器每 5 min 的新登录玩家数，但因为网络或者客户端设备故障等因素，12:00 的玩家登录日志可能在 12:10 才被收集到服务器，如果实时计算在 12:05:00 就输出结果，必然会漏掉这条迟到的数据。在离线计算中这样的问题并不明显，因为一个批次的时间跨度较大且对延迟要求不高，因此计算的时间可以设置一个安全的延迟，比如 1 个小时，确保数据都已经收集完成后再开始计算，即使有大量数据是在 1 个小时后才收集到，只需要重算结果即可。然而这样的实践经验并不能应用于实时计算，一是引入额外的安全延迟对于很多对延迟敏感的场景不可接受，二是实时计算的重算要比批处理重算的成本高出很多。因此业界普遍是采用结合离线和实时处理的 Lambda 架构来应对这个问题，其主要思想是同时运行实时和离线两个数据处理管道，实时管道提供最近小时内的临时结算结果，而离线管道提供小时以前的计算结果并覆盖掉对应时间段的实时计算结果，查询时将两者的结果再进行合并产生最终的结果 [1]。
实时计算与离线计算的融合实时计算与离线计算的分离说明了用批处理模型不足以表达流计算，于是人们开始探索批处理是流计算特例的模型。2015 年 Google 发表名为 The Dataflow Model 的论文，这篇论文较为详细地阐述了实时流计算和离线批计算的统一模型（出于篇幅原因不展开讲，详情请见 [2] ），而该模型基于批处理是流计算特例的观点。The Dataflow Model 将计算分为四个要素，即 what、where、when 和 how:

what 表示要计算什么结果，即对数据的一系列转换操作；
where 表示结果计算上下文，即窗口如何定义；
when 表示何时输出和物化计算结果；
how 表示如何清理已经输出的结果。

在 what 和 where 两点上流计算和批处理是相似的，而主要不同之处在于 when 和 how 两点，这两点在批处理里基本不会涉及，但在流计算里却影响着计算结果的准确性，实际上它们分别对应了上文所说的批处理经验不能应用于实时计算的两个问题。本文主要讨论的 watermark 就是属于 when 要素里的一种技术，因而下文将主要关注 when 。
在批处理中 when 是输入数据集结束的时候，how 是以覆盖的形式来清理之前的输出结果，处理模式都是固定的，因此用户并不需要考虑。举个例子，假设要计算一个游戏每天的玩家充值金额，用离线计算时我们会考虑如何将充值金额从日志中提取出来并累加到一起，此为 what；再考虑批处理的运行时间，比如每天 00:30，所以每次计算是处理 24 小时采集到的数据，此为 where；而批处理的 when 是和 where 绑定的，即 00:30 计算开始，结束后马上输出结果；至于 how，不同批次的批处理运行的结果是互不相干的，同一批次的运行结果会覆盖前一次运行的结果。