实时计算可以基于时间属性对数据进行窗口聚合。基于Event Time时间属性的窗口函数作业中,数据源表的声明中需要使用Watermark方法。

定义

由于实时计算的输入数据是持续不断的,因此我们需要一个有效的进度指标,来帮助我们确定关闭时间窗口的正确时间点,保证关闭窗口后不会再有数据进入该窗口,可以安全输出这个窗口的聚合结果。而Watermark就是一种衡量Event Time进展的有效机制。随着时间的推移,最早流入实时计算的数据会被处理完成,之后流入的数据处于正在处理状态。处于正在处理部分的和已处理部分的交界的时间戳,可以被定义为Watermark,代表在此之前的事件已经被处理完成并输出。

针对乱序的流,Watermark也至关重要,即使部分事件延迟到达,也不会影响窗口计算的正确性。此外,并行数据流中,当算子(Operator)有多个输入流时,算子的Event Time以最小流Event Time为准。

语法

Watermark的定义是数据源表DDL定义的一部分,Watermark语法定义如下。
WATERMARK [watermarkName] FOR <rowtime_field> AS withOffset(<rowtime_field>, offset)
参数 是否必填 说明
watermarkName 标识Watermark的名字。
<rowtime_field> <rowtime_field>必须是表中已定义的一列(当前仅支持TIMESTAMP类型),基于该列生成Watermark,并且标识该列为Event Time列。您可以使用<rowtime_field>在作业代码中定义窗口。
withOffset Watermark的生成策略,根据<rowtime_field> - offset生成Watermark的值。withOffset的第一个参数必须是<rowtime_field>
offset Watermark值与Event Time值的偏移量,单位为毫秒。

示例

通常,一条记录中的某个字段代表了该记录的发生时间。例如,表中rowtime字段的类型为TIMESTAMP,1501750584000(2017-08-03 08:56:24.000)。定义一个基于rowtime列,偏移4秒的Watermark。
WATERMARK FOR rowtime AS withOffset(rowtime, 4000)
这条数据的Watermark时间为1501750584000 - 4000 = 1501750580000(2017-08-03 08:56:20.000)。这条数据的Watermark时间含义:时间戳小于1501750580000(2017-08-03 08:56:20.000)的数据已经全部到达。
说明
  • Event TimeProcessing Time只能在源表声明。
  • 在使用Event Time Watermark时,rowtime必须是TIMESTAMP类型。如果您的数据源表中没有TIMESTAMP类型的列,可以使用计算列方法从其它类型的字段进行转换。
  • 当前实时计算支持毫秒级别的、在Unix时间戳里是13位TIMESTAMP数据类型。如果您的数据源表的rowtime字段为非13位的时间戳,可以使用计算列方法完成转化。