APPLY_WATERMARK-实时计算 Flink版(Flink)-阿里云帮助中心

在传统 Flink SQL 中，Watermark 只能在 CREATE TABLE DDL 中定义。当用户无权修改表 DDL、需要基于视图或子查询补充 Watermark，或者多个作业复用同一数据源但需要不同的 Watermark 策略时，无法灵活配置 Watermark。

为解决这些问题，VVR 提供了内置表值函数 APPLY_WATERMARK。该函数支持在查询中为表、视图或子查询定义或重新定义 Watermark；在流模式下，它会将指定的时间列标记为事件时间属性（Rowtime），并透传输入数据。生成的 Rowtime 和 Watermark 可用于窗口聚合、窗口 Join、事件时间 Interval Join 和事件时间 Temporal Join 等事件时间计算。

使用限制

仅实时计算引擎 VVR 11.9 及以上版本支持APPLY_WATERMARK函数。

语法

APPLY_WATERMARK(
  [input =>] TABLE <table_or_view_name> | (<subquery>),
  [rowtime =>] DESCRIPTOR(<rowtime_column>),
  [strategy =>] <watermark_strategy>
);

入参

参数	说明	示例
`input`	表参数（Table Argument），用于传入表、视图或子查询。	`TABLE Orders`
`rowtime`	列描述符（Column Descriptor），用于指定需要标记为 Rowtime 的单个列。该列必须为 `TIMESTAMP(p)`或`TIMESTAMP_LTZ(p)`类型，`p` 的取值范围为 0～3。	`DESCRIPTOR(order_time)`
`strategy`	标量表达式（Scalar Expression），用于定义 Watermark 生成策略。表达式的结果类型必须为 `TIMESTAMP(p)`或`TIMESTAMP_LTZ(p)`，`p` 的取值范围为 0～3。	`order_time - INTERVAL '5' SECOND`

示例

假设 Orders表包含 order_id BIGINT和order_time TIMESTAMP(3)字段。

为表定义 Watermark

以下语句将 order_time标记为 Rowtime，并生成相对 order_time滞后 5 秒的 Watermark：

SELECT
  order_id,
  order_time
FROM APPLY_WATERMARK(
  TABLE Orders,
  DESCRIPTOR(order_time),
  order_time - INTERVAL '5' SECOND
);

为子查询定义 Watermark

input参数支持传入子查询。子查询需要输出 Rowtime 列以及 Watermark 策略引用的列：

SELECT
  order_id,
  order_time
FROM APPLY_WATERMARK(
  (
    SELECT
      order_id,
      order_time
    FROM Orders
    WHERE order_time IS NOT NULL
  ),
  DESCRIPTOR(order_time),
  order_time - INTERVAL '5' SECOND
);

使用命名参数

以下示例通过参数名传入参数，并支持调整参数顺序：

SELECT
  order_id,
  order_time
FROM APPLY_WATERMARK(
  strategy => order_time - INTERVAL '5' SECOND,
  rowtime => DESCRIPTOR(order_time),
  input => TABLE Orders
);

用于窗口聚合

如果需要将结果用于窗口表值函数，请先创建视图：

CREATE TEMPORARY VIEW WatermarkedOrders AS
SELECT
  order_id,
  order_time
FROM APPLY_WATERMARK(
  TABLE Orders,
  DESCRIPTOR(order_time),
  order_time - INTERVAL '5' SECOND
);

再基于该视图进行窗口聚合：

SELECT
  window_start,
  window_end,
  COUNT(*) AS order_count
FROM TABLE(
  TUMBLE(
    TABLE WatermarkedOrders,
    DESCRIPTOR(order_time),
    INTERVAL '1' MINUTE
  )
)
GROUP BY window_start, window_end;

重新定义 Watermark 生成策略

假设 OrdersWithWatermark已定义滞后 10 秒的 Watermark。以下 DDL 仅展示表结构和原 Watermark，连接器选项请根据实际 Source 配置：

CREATE TABLE OrdersWithWatermark (
  order_id BIGINT,
  order_time TIMESTAMP(3),
  WATERMARK FOR order_time AS order_time - INTERVAL '10' SECOND
);

可以使用 APPLY_WATERMARK重新定义 Watermark 生成策略：

SELECT
  order_id,
  order_time
FROM APPLY_WATERMARK(
  TABLE OrdersWithWatermark,
  DESCRIPTOR(order_time),
  order_time - INTERVAL '5' SECOND
);

上游原有的 Watermark 节点可能仍会保留，但最外层 APPLY_WATERMARK的策略决定向下游发出的普通 Watermark 时间戳。

级联 APPLY_WATERMARK

APPLY_WATERMARK支持级联调用。以下示例先定义滞后 10 秒的 Watermark，再在外层重新定义为滞后 5 秒：

SELECT
  order_id,
  order_time
FROM APPLY_WATERMARK(
  (
    SELECT
      order_id,
      order_time
    FROM APPLY_WATERMARK(
      TABLE Orders,
      DESCRIPTOR(order_time),
      order_time - INTERVAL '10' SECOND
    )
  ),
  DESCRIPTOR(order_time),
  order_time - INTERVAL '5' SECOND
);

级联调用时，最外层（最靠近下游）的 APPLY_WATERMARK策略决定该数据分支向下游发出的普通 Watermark 时间戳。