关于撤回流的Watermark问题

[lec ssmi]

Hi:
   在tableAPI中，带有时间属性的聚合，比如window聚合，对于retract消息的事件延迟怎么处理呢？
   举个例子，
   假设上游使用了last_value 操作加over window操作，一直生成一条数据的最新值，然后和另外一个流做join , 再进行 time
window聚合操作。现在已经十点，最大延迟为一个小时，这个时候event-time为9点的消息，已经超过了最大延迟,但是在join中仍然生成一条join后的记录(因为join不过滤超时的数据)，这条记录会对原先join好的一条记录进行撤回。那这个撤回消息，在到达time
window的时候，因为超过了最大延迟，DELELE记录和INSERT记录都不会被处理吗？
 谢谢。

[Benchao Li]

Hi lec，

Window Operator目前是不支持retract的输入的。

lec ssmi <[hidden email]> 于2020年4月28日周二 上午9:45写道：

> Hi:
>    在tableAPI中，带有时间属性的聚合，比如window聚合，对于retract消息的事件延迟怎么处理呢？
>    举个例子，
>    假设上游使用了last_value 操作加over window操作，一直生成一条数据的最新值，然后和另外一个流做join , 再进行 time
>
> window聚合操作。现在已经十点，最大延迟为一个小时，这个时候event-time为9点的消息，已经超过了最大延迟,但是在join中仍然生成一条join后的记录(因为join不过滤超时的数据)，这条记录会对原先join好的一条记录进行撤回。那这个撤回消息，在到达time
> window的时候，因为超过了最大延迟，DELELE记录和INSERT记录都不会被处理吗？
>  谢谢。
>


--

Benchao Li
School of Electronics Engineering and Computer Science, Peking University
Tel:[phone]
Email: [hidden email]; [hidden email]

[lec ssmi]

但是我在DataStreamGroupWindowAggregateBase这个类里面，发现以下两个方法都是true啊。
override def needsUpdatesAsRetraction = true
override def consumesRetractions = true

Benchao Li <[hidden email]> 于2020年4月28日周二 上午10:19写道：

> Hi lec，
>
> Window Operator目前是不支持retract的输入的。
>
> lec ssmi <[hidden email]> 于2020年4月28日周二 上午9:45写道：
>
> > Hi:
> >    在tableAPI中，带有时间属性的聚合，比如window聚合，对于retract消息的事件延迟怎么处理呢？
> >    举个例子，
> >    假设上游使用了last_value 操作加over window操作，一直生成一条数据的最新值，然后和另外一个流做join , 再进行
> time
> >
> >
> window聚合操作。现在已经十点，最大延迟为一个小时，这个时候event-time为9点的消息，已经超过了最大延迟,但是在join中仍然生成一条join后的记录(因为join不过滤超时的数据)，这条记录会对原先join好的一条记录进行撤回。那这个撤回消息，在到达time
> > window的时候，因为超过了最大延迟，DELELE记录和INSERT记录都不会被处理吗？
> >  谢谢。
> >
>
>
> --
>
> Benchao Li
> School of Electronics Engineering and Computer Science, Peking University
> Tel:[phone]
> Email: [hidden email]; [hidden email]
>

[Benchao Li]

Hi lec,

不好意思，这个行为是之前的行为。这个最近已经在FLINK-16887中顺手把这个行为修改了。

我之前也建了一个issue：https://issues.apache.org/jira/browse/FLINK-16844 想修改这个行为的，
不过还没有得到committer的认同。这个issue那我现在可以关掉了。

那么我们按照当前最新的代码来看的话，你可以认为Window Operator是支持retract消息的。
那么对于retract消息的处理，其实跟普通的append消息是相同的，过期的retract消息同样会被直接丢弃。



lec ssmi <[hidden email]> 于2020年4月28日周二 上午11:28写道：

> 但是我在DataStreamGroupWindowAggregateBase这个类里面，发现以下两个方法都是true啊。
> override def needsUpdatesAsRetraction = true
> override def consumesRetractions = true
>
> Benchao Li <[hidden email]> 于2020年4月28日周二 上午10:19写道：
>
> > Hi lec，
> >
> > Window Operator目前是不支持retract的输入的。
> >
> > lec ssmi <[hidden email]> 于2020年4月28日周二 上午9:45写道：
> >
> > > Hi:
> > >    在tableAPI中，带有时间属性的聚合，比如window聚合，对于retract消息的事件延迟怎么处理呢？
> > >    举个例子，
> > >    假设上游使用了last_value 操作加over window操作，一直生成一条数据的最新值，然后和另外一个流做join , 再进行
> > time
> > >
> > >
> >
> window聚合操作。现在已经十点，最大延迟为一个小时，这个时候event-time为9点的消息，已经超过了最大延迟,但是在join中仍然生成一条join后的记录(因为join不过滤超时的数据)，这条记录会对原先join好的一条记录进行撤回。那这个撤回消息，在到达time
> > > window的时候，因为超过了最大延迟，DELELE记录和INSERT记录都不会被处理吗？
> > >  谢谢。
> > >
> >
> >
> > --
> >
> > Benchao Li
> > School of Electronics Engineering and Computer Science, Peking University
> > Tel:[phone]
> > Email: [hidden email]; [hidden email]
> >
>

--

Benchao Li
School of Electronics Engineering and Computer Science, Peking University
Tel:[phone]
Email: [hidden email]; [hidden email]