「吉光片羽：文献阅读记录」AdaptiveWatermarks:AConceptDriftbasedApproachforPredictingEventTimeProgress

前言：

还有一年半毕业，准备回学校搞论文，虽然不知道能不能搞出来，但还是得试试，年前看的论文内容都快忘了，开一个坑，边看边记录，给自己和后人一点痕迹。主要是翻译论文内容，不会带有自己的想法，会略去口水话，做到尽量简洁。

阅读者最好有一定的流数据处理基础（应该没人看吧）

本篇是德国几个大学联合写的一个短篇，发布于Advances in Database Technology - EDBT2019-March，被引用7次，是low-watermark方面为数不多的文章，且看且珍惜。

本文提出了一种自适应水印生成策略并在Apache Flink中实现。
我们的策略可以自适应地决定何时生成watermark以及使用什么时间戳，而无需预先调整。我们使用一个自适应窗口(adaptive window-ADWIN)来检测流数据中的eventtime变化并提高到达率和降低延迟。
两个真实数据集的实验结果表明，我们的策略通过提前触发窗口实现了较低的平均延迟，并且在预期无序数据时通过延迟水印实现了较低的丢弃率。

术语一览

由于通常数据源是分布式的，而且可能远离SPE，因此元素e在到达SPE前可能会出现延迟和无序。既，两个元素e1和e2，其中te(e1)tp(e2)，这就是所谓的无序到达。

数据流通常被一系列window所切割成明确大小的块来进行计算，如求一段时间内的平均值。

Time window是一种常见的window类型，它根据时间进展来划分数据流，既可以根据SPE内部时钟来划分，也可以根据外部数据自带的时间戳来进行处理。在后一种方法中，我们需要引入一些外部的概念。

low watermark（水印）是一种考虑事件时间的技术，表示后续不会出现携带比当前水印更早的时间戳的数据。流中的window操作符接收到水印时会触发完成时间窗口的执行。

水印的一些特点：1、水印是单调的。2、当生成的水印太少，计算结果的时效性无法保证。3、当生成的水印太多，会丢失更多的数据。

当前水印生成方法有两种，一种是启发式的和周期性的，另一种是punctuation-based(实在不知道咋翻译),前者对数据到达率的变化或数据延迟的变化不灵活。

在本文中，我们在周期性水印的基础上提出了一种可以灵活变化且无需提前知道数据内容的水印生成方法，我们称之为adaptive window（ADWIN），来决定合适生成新水印以及用哪个值生成新水印。此外还提供了一种控制方法——延迟到达阈值。

我们在Flink上实现了我们的方法并进行了比较，并通过实验证明了我们的方法在减少延迟和减少数据丢失方面的优越性。

首先，我们要研究两个东西，m——最大乱序时间，s——生成水印的周期。当流中的新数据源源不断地到来时，我们要不断更新这两个值。当自上一次生成水印以来，迟到数据个数/总数据个数的值超过某个阈值 L 时，产生新的水印。

这样一来，水印产生的间隔会根据数据到达延迟的变化而变化(adaptive)，同样的，m也可以在每次变化时更新。为了检测到达间隔时间的变化，我们使用ADWIN算法。

ADWIN的工作原理是随着时间的推移维护一个window内的数据集合，窗口大小随着输入数据变化频率的变化而变化。数据流发生的变化约大，窗口大小越小。该算法的参数表如下。

δ越高，系统对变化越敏感，初始设为1，这样系统能更早地探测到变化。

ADWIN工作原理如下：

————————————————————————————————————————————————————————————————————
算法 1 ： Adaptive watermark generation
____________________________________________________________________
输入：数据流S
输入：敏感度变化率∆δ
输入：迟到阈值l
输入：初始化用到的数据数量warmup
warmup=0;m=0;watermark=0;
lateElements=0;totalelements=0;δ=1;
maxTimeStamps=-∞
adWin = initializeAdwin(δ)
foreach e∈S do
maxTimestamp = max(te(e),maxTimestamp) //更新检测到的最大事件时间
if warmup ≤ w then //这里处理初始化用的数据
m=max(m,tp(e)-te(e)) //找出预热数据中的最大 摄入时间和事件时间 之差，记为m
warmup=warmup+1
else //这里处理正式数据
totalElements = totalElements + 1 //计数器加一
if adWin.driftDetected((tp(e)-te(e))/m,δ) //用(摄入时间-事件事件)/m和δ来检测是否发生概念漂移，若发生了执行下面
then
if lateElements = 0 then //如果当前无迟到数据，灵敏度低，要提高敏感度
δ = increaseSensitivity(∆δ)
if lateElements/totalElements watermark = maxTimestamp - m
emit(watermark)
lateElements=0
totalElements=0
else //如果比值大于等于l，此时过于灵敏，需要降低敏感度
m=updateSkewness()
δ=decreaseSensitivity(∆δ)
else //如果没有发生概念漂移
if te(e) lateElements=lateElements+1

新元素到达时，将新元素的事件时间戳与摄入时间之差被插入ADWIN，并进行检查来检测变化。这样一个检测可以作为是否生成新水印的一个指标，我们没必要每次插入一个数据都生成新水印，我们还要看第二个指标迟到率，只有检测变化时，迟到率低于检测阈值l时才会生成新水印，每次生成新水印时重置此速率。阈值l决定了程序的变化灵敏度，如果需要更敏锐的灵敏度，就把l设的低一些。

下面我们用图示和文字结合起来演示一下效果

上图举例说明了本算法的原理，并与假想的固定周期水印生成策略（s=3

秒，m=5秒）进行了比较

Adaptive watermark策略中的参数：l=0.5，∆δ=1，w=3

垂直线代表流数据的摄入时间和事件时间

固定周期生成策略会在tp=103、106、109处分别生成值为95、98、101的水印，该策略中tp=105、107、109的数据迟到了。

AW算法在tp=100和102之间进行初始化，可以观察到在tp=103时检测到了概念漂移并生成一个水印，tp=106时同样的生成一个值为97的新水印。tp=107的是迟到数据。tp=108时检测到概念漂移，没有水印产生。在tp=110时，检测到概念漂移并产生值为98的水印。

后面的实验结果就不写了，结果当然很棒棒啦