【Flink】Watermark 在 Flink SQL 中的用法

欢迎来到我的博客，很高兴能够在这里和您见面！欢迎订阅相关专栏：
⭐️ 全网最全IT互联网公司面试宝典：收集整理全网各大IT互联网公司技术、项目、HR面试真题.
⭐️ AIGC时代的创新与未来：详细讲解AIGC的概念、核心技术、应用领域等内容。
⭐️ 大数据平台建设指南：全面讲解从数据采集到数据可视化的整个过程，掌握构建现代化数据平台的核心技术和方法。
⭐️《遇见Python：初识、了解与热恋》 ：涵盖了Python学习的基础知识、进阶技巧和实际应用案例，帮助读者从零开始逐步掌握Python的各个方面，并最终能够进行项目开发和解决实际问题。
⭐️《MySQL全面指南：从基础到精通》通过丰富的实例和实践经验分享，带领你从数据库的基本操作入手，逐步迈向复杂的应用场景，最终成为数据库领域的专家。
⭐️ 数据治理：通过通俗易懂的文章，学者们不仅能理解数据治理的重要性，还能掌握数据治理的基本原则和最佳实践。

在 Flink SQL 中，Watermark 的使用稍有不同。你可以通过在定义表的 DDL（Data Definition Language）中指定 Watermark 来处理乱序数据。在 Flink SQL 中，Watermark 是与事件时间字段相关联的，它用于告诉 Flink 处理乱序数据的策略。

1. 定义 Watermark 在 Flink SQL 中的基础语法

在 Flink SQL 中，定义 Watermark 的方式是在创建表时，使用

WATERMARK FOR

语句。它通常跟在事件时间列的定义后面，指定如何从事件时间生成 Watermark。语法如下：

CREATETABLE your_table (
    column1 TYPE,
    event_time_column TIMESTAMP(3),
    WATERMARK FOR event_time_column AS watermark_strategy_expression
)WITH('connector'='...',-- 例如 'kafka''topic'='your_topic',...);

在这里：

• event_time_column 是事件时间列。
• watermark_strategy_expression 是 Watermark 生成策略的表达式。

2. Watermark 策略

2.1 固定延迟的 Watermark

最常用的 Watermark 策略是固定延迟策略。你可以通过指定事件时间减去一定的延迟（例如 5 秒）来生成 Watermark。

WATERMARK FOR event_time_column AS event_time_column -INTERVAL'5'SECOND

2.2 自定义 Watermark 生成策略

你也可以定义更复杂的 Watermark 生成策略，例如基于业务逻辑生成 Watermark。下面是一个自定义的例子，延迟 10 分钟的 Watermark：

CREATETABLE orders (
    order_id STRING,
    user_id STRING,
    amount DOUBLE,
    order_time TIMESTAMP(3),
    WATERMARK FOR order_time AS order_time -INTERVAL'10'MINUTE)WITH('connector'='kafka','topic'='orders_topic','properties.bootstrap.servers'='localhost:9092','format'='json');

在上面的例子中，

WATERMARK FOR order_time AS order_time - INTERVAL '10' MINUTE

指定了 Watermark 的生成策略：假设事件时间延迟最多 10 分钟到达。

3. 使用 Watermark 的实际 SQL 查询

一旦表定义了 Watermark，后续的 SQL 查询可以基于 Watermark 执行窗口聚合、时间窗口等操作。下面是一个使用 Watermark 的示例查询，计算每 1 分钟的订单金额总和：

SELECT 
  TUMBLE_START(order_time,INTERVAL'1'MINUTE)AS window_start,
  TUMBLE_END(order_time,INTERVAL'1'MINUTE)AS window_end,SUM(amount)AS total_amount
FROM orders
GROUPBY TUMBLE(order_time,INTERVAL'1'MINUTE);

在这个查询中，

TUMBLE(order_time, INTERVAL '1' MINUTE)

基于 Watermark 处理 1 分钟时间窗口内的订单数据。

4. 常见 Watermark 策略

• **Bounded Out-of-Orderness (有界乱序)**：允许事件数据乱序一定时间到达。WATERMARK FOR event_time AS event_time -INTERVAL'5'SECOND
• **Monotonous (单调递增)**：假设事件时间是单调递增的，不允许乱序。WATERMARK FOR event_time AS event_time

5. 示例案例：处理传感器数据

假设我们有一个传感器数据流，每个传感器定期发送温度数据，数据的时间戳可能会有一些延迟。我们使用 Watermark 来应对延迟数据，并统计每 10 秒内的平均温度。

表定义：

CREATETABLE sensor_data (
    sensor_id STRING,
    temperature DOUBLE,
    ts TIMESTAMP(3),
    WATERMARK FOR ts AS ts -INTERVAL'5'SECOND)WITH('connector'='kafka','topic'='sensor_topic','properties.bootstrap.servers'='localhost:9092','format'='json');

查询：

SELECT
    TUMBLE_START(ts,INTERVAL'10'SECOND)AS window_start,
    TUMBLE_END(ts,INTERVAL'10'SECOND)AS window_end,AVG(temperature)AS avg_temperature
FROM sensor_data
GROUPBY TUMBLE(ts,INTERVAL'10'SECOND);

在这个例子中，

WATERMARK FOR ts AS ts - INTERVAL '5' SECOND

表示允许数据最多延迟 5 秒到达，并在每 10 秒的时间窗口内计算平均温度。

6. 结论

Flink SQL 中的 Watermark 机制简化了对无序数据的处理。通过定义 Watermark，Flink 可以基于事件时间准确地处理乱序数据，确保数据分析结果的准确性。Watermark 的灵活性允许你根据不同场景定制延迟策略，适应现实数据流的复杂性。

💗💗💗💗💗💗💗💗💗💗💗💗

💗💗💗💗💗💗💗💗💗💗💗💗