Watermark

watermark 是flink处理数据延迟及乱序的手段，具体细节参考flink中的的watermark的详细设计及作用

如何在sql中使用watermark

DDL任务

CREATE TABLE user_actions (
  user_name STRING,
  data STRING,
  user_action_time TIMESTAMP(3),
  -- declare user_action_time as event time attribute and use 5 seconds delayed watermark strategy
  WATERMARK FOR user_action_time AS user_action_time - INTERVAL '5' SECOND
) WITH (
  ...
);

SELECT TUMBLE_START(user_action_time, INTERVAL '10' MINUTE), COUNT(DISTINCT user_name)
FROM user_actions
GROUP BY TUMBLE(user_action_time, INTERVAL '10' MINUTE);

在ddl中添加watermark定义的字段即可使用watermark

Metahub任务

SET table_name.SCHEMA.WATERMARK.field_name = 'watermark_strategy_expression';
-- table_name是定义的流表名
-- field_name 是流表中字段名称，字段类型必须是timestamp(3)
-- watermark_strategy_expression 是watermark定义表达式，例如field_name - INTERVAL '5' SECOND

如果流表没有timestamp(3)类型的字段，又需要将该字段定义为event_time，需要使用计算列方式转为timestamp(3)。

SET 'table_name.COMPUTED.FIELD.EXPRESSION.tsudf' = 'GetTs(ts_str)';
-- table_name是定义的流表名
-- ts_str是流表中字段名称，通过GetTs udf函数返回tsudf字段
-- tsudf是计算后返回的字段名
SET 'table_name.COMPUTED.FIELD.TYPE.tsudf' = 'TIMESTAMP';
-- 定义tsudf字段类型为timestamp。

SET table_name.SCHEMA.WATERMARK.tsudf = 'tsudf - INTERVAL '5' SECOND';
-- 定义watermark 允许延迟5秒钟。

create function GetTs as 'com.netease.udf.GetTsColumns';
-- 定义udf函数，传入string字段，返回timestamp类型字段。

----以上sql相当于执行了：
create table table_name (
    ts_str string,
    tsudf as GetTs(ts_str)
) WITH (...);

create function GetTs as 'com.netease.udf.GetTsColumns';