实时计算大数据处理的基石-Google Dataflow

此文选自google大神tyler akidau的另一篇文章：streaming 102: the world beyond batch

欢迎回来！如果您错过了我以前的帖子，Streaming-大数据的未来，强烈建议您先花时间阅读那篇文章。

简要回顾一下，上一篇我们介绍了Streaming，批量与流式计算，正确性与推理时间的工具，数据处理模式，事件事件与处理时间，窗口化。

在这篇文章中，我想进一步关注上次的数据处理模式，但更详细。

这里会用到一些Google Cloud Dataflow[1]的代码片段，这是谷歌的一个框架，类似于Spark Streaming或Storm。

这里还有再说三个概念：

Watermarks：水印是关于事件时间的输入完整性的概念。如果到某一个时间的水印，应该是已经获取到了小于该时间的所有数据。在处理无界数据时，水印就作为处理进度的标准。

Triggers: 触发器是一种机制，用于声明窗口何时应该输出，触发器可灵活选择何时应发出输出。我们可以随着时间的推移不断改进结果，也可以处理那些比水印晚到达的数据，改进结果。

Accumulation: 累积模式指定在同一窗口中观察到的多个结果之间的关系。这些结果可能是完全脱节的，即随着时间的推移表示独立的增量，或者它们之间可能存在重叠。

四个新的问题：what？where？when？How？

计算什么？ 希望通过数据计算的结果，和批处理类似，构建直方图，计算总和，训练机器学习等等。

在哪里计算？事件时间窗口可以回答这个问题，比如之前提到的（固定，滑动，会话），当然这个时间也可能是处理时间。

什么时候处理产生结果？通过水印和触发器来回答。可能有无限的变化，常见的模式是使用水印描述给定窗口的输入是否完整，触发器指定早期和后期结果。

结果如何相关？ 通过累计模式来回答，丢弃不同的，累积产生的结果。

详细介绍Streaming 101的一些概念，并提供一些例子。

计算的结果是什么？熟悉批处理的应该很熟悉这个。

举一个例子，计算由10个值组成的简单数据集的整数和。您可以想象为求一组人的分数和，或者是计费，监控等场景。

如果您了解Spark Streaming或Flink之类的东西，那么您应该相对容易地了解Dataflow代码正在做什么。

Dataflow Java SDK 模型:

图二 转换类型

我们从IO源中获取消息，以KV的形式转换，最后求出分数和。示例代码如下：

PCollection raw = IO.read(...);PCollection> input = raw.apply(ParDo.of(new ParseFn());PCollection> scores = input  .apply(Sum.integersPerKey());

这个过程可以是在多个机器分布式执行的，分布的将不同时间情况的数据进行累加，输出得到最终的结果，我们不用关心分布式的问题，只要把所有的结果集转换累加即可。

图三 x为事件时间 y为处理时间

这里我们计算的是所有事件时间，没有进行窗口转换，因此输出矩形覆盖整个X轴，但是我们处理无界数据时，这就不够了，我们不能等到结束了再处理，因为永远不会结束。所有我们需要考虑在哪里计算呢？这就需要窗口。

还记得我们之前提过的三种窗口，固定，滑动，会话。

图四 三种窗口

我们用刚才的例子，将其固定为两分钟的窗口。

PCollection> scores = input  .apply(Window.into(FixedWindows.of(Duration.standardMinutes(2))))  .apply(Sum.integersPerKey());

Dataflow提供了一个统一的模型，可以在批处理和流式处理中同时工作，因为批处理实际上只是流的一个子集。

图五 窗口处理

和以前一样，输入的数据在累积，直到它们被完全处理，然后产生输出。在这种情况下，我们得到四个输出而不是一个输出：四个基于这个两分钟事件时间窗口中的单个输出。

现在我们可以通过更具体的水印，触发器和累计来解决更多的问题了。

刚才的处理还是通用的批处理方式，延迟很大，但我们已经成功把每个窗口的输入都计算了，我们目前缺乏一种对无限数据处理方法，还要能保证其完整性。

水印是什么时候处理产生结果？其实也就是我们之前研究事件时间和处理时间的那张图。

上文图 事件时间 处理时间 水印

这条红色曲线就是水印，它随着处理时间的推移不断的去捕获事件时间。从概念上讲，我们将其视为从处理时间到事件时间的映射。水印可以有两种类型：

完美水印：这要求我们对的输入数据全部了解。也就没有了后期数据，所有的数据准时到达。

启发式水印：对于大部分分布式输入源，完整的了解输入数据是不可能的，这就需要启发式水印。启发式水印通过分区，分区排序等提供尽可能准确的估计。所以是有可能错误的，这就需要触发器在后期解决，这个一会会讲。

下面是两个使用了不同水印的流处理引擎：

图六 左完美 右启发

在这两种情况下，当水印通过窗口的末端时，窗口被实现。两次执行之间的主要区别在于右侧水印计算中使用的启发式算法未考虑9的值，这极大地改变了水印的形状。这些例子突出了水印的两个缺点：

太慢：如果因为网络等原因导致有数据未处理时，只能延迟输出结果。左图比较明显，迟到的9影响了整体的进度，这对于第二个窗口[12：02,12：04]尤为明显，从窗口中的第一个值开始到我们看到窗口的任何结果为止需要将近7分钟。而启发式水印要好一点只用了两分钟。

太快：当启发式水印错误地提前超过应有的水平时，水印之前的事件时间数据可能会在一段时间后到达，从而产生延迟数据。这就是右边示例中发生的情况：在观察到该窗口的所有输入数据之前，水印超过了第一个窗口的末尾，导致输出值不正确，正确的应该是14。这个缺点严格来说是启发式水印的问题， 他们的启发性意味着他们有时会出错。因此，如果您关心正确性，单靠它们来确定何时实现输出是不够的。

这时候我们就需要触发器。

触发器用于声明窗口何时应该输出。

触发的信号包括：水印进度，处理时间进度，计数，数据触发，重复，逻辑与AND，逻辑或OR，序列。

还是用上面的例子，我们增加一个触发器：

PCollection> scores = input  .apply(Window.into(FixedWindows.of(Duration.standardMinutes(2)))               .triggering(AtWatermark()))  .apply(Sum.integersPerKey());

这里规定了触发的情况，我们可以考虑水印太快和太慢的情况。

太慢时，我们假设任何给定窗口都存在稳定的传入，我们可以周期性的触发。

太快时，可以在后期数据到达后去修正结果。如果后期数据不频繁，并不会影响性能。

最后我们可以综合考虑，协调早期，准时，晚期的情况：

PCollection> scores = input  .apply(Window.into(FixedWindows.of(Duration.standardMinutes(2)))               .triggering(                 AtWatermark()                   .withEarlyFirings(AtPeriod(Duration.standardMinutes(1)))                   .withLateFirings(AtCount(1))))  .apply(Sum.integersPerKey());

生成结果如下，这个版本有了明显的改进：

图七 增加早期晚期

对于[12：02,12：04]窗口太慢的情况，每分钟定时更新。延迟时间从七分钟减少到三分半。

对于[12：00,12：02]窗口太快的情况，当值9显示较晚时，我们立即将其合并到一个值为14的新的已更正窗格中。

但是这里有一个问题，窗口要保持多长时间呢？这里我们需要垃圾收集机制。

在[启发式水印示例中，每个窗口的持久状态在示例的整个生命周期，这是必要的，这样我们才能够在他们到达时适当处理迟到的数据。但是，虽然能够保持所有持久状态直到时间结束是很棒的，但实际上，在处理无限数据源时，保持给定窗口的状态通常是不切实际的。无限， 我们最终会耗尽磁盘空间。

因此，任何真实的无序处理系统都需要提供一些方法来限制它正在处理的窗口的生命周期。

我们可以定义一个范围，当超出这个范围后，我们就丢弃无用的数据。

PCollection> scores = input  .apply(Window.into(FixedWindows.of(Duration.standardMinutes(2)))               .triggering(                 AtWatermark()                   .withEarlyFirings(AtPeriod(Duration.standardMinutes(1)))                   .withLateFirings(AtCount(1)))               .withAllowedLateness(Duration.standardMinutes(1)))  .apply(Sum.integersPerKey());

一旦水印通过窗口的延迟范围，该窗口就会关闭，这意味着窗口的所有状态都将被丢弃。

图八 垃圾收集

这里的6在允许迟到的范围内，可以被收集，而9不在这个范围，就被丢弃了。

有两点要注意：

如果您正在使用可获得完美水印的数据源的数据，就不需要处理延迟数据。

即使在使用启发式水印时，如果是将有限数量聚合，而且能保证一直可控，也不用考虑窗口的寿命问题。

现在时间的问题解决了，下面我们讨论如何累积数据。

有三种不同的累积模式：

丢弃：当下游的消费者进行累积计算时，直接相加所要的，就可以得到最终结果。

华友协同办公自动化OA系统

华友协同办公管理系统(华友OA)，基于微软最新的.net 2.0平台和SQL Server数据库，集成强大的Ajax技术，采用多层分布式架构，实现统一办公平台，功能强大、价格便宜，是适用于企事业单位的通用型网络协同办公系统。 系统秉承协同办公的思想，集成即时通讯、日记管理、通知管理、邮件管理、新闻、考勤管理、短信管理、个人文件柜、日程安排、工作计划、工作日清、通讯录、公文流转、论坛、在线调查、

下载

累积：比如未来的可以覆盖之前的，一直要保持最新状态，例如Hbase这种键值对的存储。

累积和撤回：和累积类似，但更复杂。比如重新分组的情况，可能不只是覆盖那么简单，需要先删掉之前的，再加入最新的；还有动态窗口的情况，新窗口会替换旧窗口，但数据要放在不同的位置。

比如上图中事件时间范围[12：02,12：04]，下表显示了三种累积模式：

丢弃

累积

累积和收回

窗格1：[7]

7

7

7

第2页：[3,4]

7

14

14，-7

第3页：[8]

8

22

22，-14

观察到最后的价值

8

22

22

总和

22

51

22

丢弃：每个窗格仅包含在该特定窗格期间到达的值。因此，观察到的最终值并未完全捕获总和。但是，如果您要自己对所有独立窗格求和，那么您将得到22的正确答案。

累积：每个窗格结合了特定窗格期间到达的值，加上从先前的窗格中的所有值。因此，正确观察到的最终值可以捕获22的总和。

累积和撤回：每个窗格都包含新的累积模式值以及前一个窗格值的缩进。因此，观察到的最后一个（非回缩）值以及所有物化窗格的总和（包括撤回）都为您提供了22的正确答案。这就是撤回如此强大的原因。

图九 三种累积模式

随着丢弃，累积，累积和撤回的顺序，存储和计算成本在提高，因此累积模式的选择要在正确性，延迟和成本中做出选择。

我们已经解决了所有四个问题，What，Where，When，How。但我们都是再事件时间的固定窗口。

所以我们还要讨论一下处理时间中的固定窗口和事件时间中的会话窗口。

先讨论处理时间中的固定窗口，处理时间窗口很重要，原因有两个：

有两种方法可用于实现处理时窗口：

触发器：忽略事件时间（即，使用跨越所有事件时间的全局窗口）并使用触发器在处理时间轴上提供该窗口的快照。

入口时间：将入口时间指定为数据到达时的事件时间，并使用正常的事件时间窗口。这基本上就像Spark Streaming目前所做的那样。

处理时间窗口的一个重大缺点是，当输入的观察顺序发生变化时，窗口的内容会发生变化。为了以更具体的方式展示，我们将看看这三个用例：

这里我们将两种事件时间相同而处理时间不同的情况比较。

事件时间窗口

图10 事件时间窗口

四个窗口最终结果依然相同。

通过触发器处理时间窗口

使用全局事件时间窗口，在处理时间域定期触发，使用丢弃模式进行

图11 触发器处理时间窗口

通过入口时间处理时间窗口

当元素到达时，它们的事件时间需要在入口时被覆盖。返回使用标准的固定事件时间窗口。由于入口时间提供了计算完美水印的能力，我们可以使用默认触发器，在这种情况下，当水印通过窗口末端时，它会隐式触发一次。由于每个窗口只有一个输出，因此累积模式无关紧要。

图12 入口时间处理时间窗口

如果您关心事件实际发生的时间，您必须使用事件时间窗口，否则您的结果将毫无意义。

动态的，数据驱动的窗口，称为会话。

会话是一种特殊类型的窗口，它捕获数据中的一段活动，它们在数据分析中特别有用。

图13 会话

我们来构建一个会话：

PCollection> scores = input  .apply(Window.into(Sessions.withGapDuration(Duration.standardMinutes(1)))               .triggering(                 AtWatermark()                   .withEarlyFirings(AtPeriod(Duration.standardMinutes(1)))                   .withLateFirings(AtCount(1)))               .accumulatingAndRetractingFiredPanes())  .apply(Sum.integersPerKey());

我们得到结果如下：

图14 会话窗口

当遇到值为5的第一个记录时，它被放置在一个原始会话窗口中。

到达的第二个记录是7，它同样被放入它自己的原始会话窗口，因为它不与5的窗口重叠。

同时，水印已经过了第一个窗口的末尾，所以5的值在12:06之前被实现为准时结果。此后不久，第二个窗口也被实现为具有值7的推测结果，正如处理时间达到12:06那样。

我们接下来观察一系列记录，3,4和3，原始会话都重叠。结果，它们全部合并在一起，并且在12:07触发的早期触发时，发出值为10的单个窗口。

当8在此后不久到达时，它与具有值7的原始会话和具有值10的会话重叠。因此所有三个被合并在一起，形成具有值25的新组合会话。

当9到达时，将值为5的原始会话和值为25的会话加入到值为39的单个较大会话中。

这个非常强大的功能，Spark Streaming[2]已经做了实现。

简单回顾一下,我们讨论了事件时间与处理时间，窗口化，水印，触发器，累积。探索了What，When，Where，How四个问题。而最终，我们将平衡正确性，延迟和成本问题，得到最适合自己的实时流式处理方案。