Flink SQL实战：写SQL就能搞定实时计算

传统数据库：表是静态的，查询返回固定结果
Flink SQL：表是动态的，查询持续运行，结果持续更新

Kafka Topic "orders" → 动态表 orders → 持续查询 → 结果动态表 → 输出

-- 创建动态表（对应Kafka Topic）
CREATE TABLE orders (
    order_id STRING,
    user_id STRING,
    amount DECIMAL(10,2),
    order_time TIMESTAMP(3),
    WATERMARK FOR order_time AS order_time - INTERVAL '5' SECOND
) WITH (
    'connector' = 'kafka',
    'topic' = 'orders',
    'properties.bootstrap.servers' = 'kafka:9092',
    'format' = 'json'
);

-- 持续查询：每分钟每用户的订单总额
SELECT
    user_id,
    TUMBLE_START(order_time, INTERVAL '1' MINUTE) AS window_start,
    SUM(amount) AS total_amount,
    COUNT(*) AS order_count
FROM orders
GROUP BY
    user_id,
    TUMBLE(order_time, INTERVAL '1' MINUTE);

-- Event Time（最常用）
CREATE TABLE events (
    event_id STRING,
    event_time TIMESTAMP(3),
    -- 声明Event Time + Watermark
    WATERMARK FOR event_time AS event_time - INTERVAL '10' SECOND
) WITH (...);

-- Processing Time
CREATE TABLE events (
    event_id STRING,
    proc_time AS PROCTIME()  -- 虚拟列
) WITH (...);

高频SQL场景

场景一：实时TopN

-- 每分钟销量Top3的商品
SELECT *
FROM (
    SELECT
        product_id,
        window_start,
        sales_count,
        ROW_NUMBER() OVER (PARTITION BY window_start ORDER BY sales_count DESC) AS rn
    FROM (
        SELECT
            product_id,
            TUMBLE_START(order_time, INTERVAL '1' MINUTE) AS window_start,
            COUNT(*) AS sales_count
        FROM orders
        GROUP BY product_id, TUMBLE(order_time, INTERVAL '1' MINUTE)
    )
)
WHERE rn <= 3;

场景二：实时去重

-- 按user_id去重，只保留第一条
SELECT user_id, event_type, event_time
FROM (
    SELECT *,
        ROW_NUMBER() OVER (PARTITION BY user_id ORDER BY event_time) AS rn
    FROM events
)
WHERE rn = 1;

场景三：写入Doris

CREATE TABLE doris_sink (
    user_id STRING,
    window_start TIMESTAMP(3),
    total_amount DECIMAL(10,2),
    PRIMARY KEY (user_id, window_start) NOT ENFORCED
) WITH (
    'connector' = 'doris',
    'fenodes' = 'doris-fe:8030',
    'table.identifier' = 'db.user_order_1min',
    'username' = 'root',
    'password' = '',
    'sink.properties.format' = 'json'
);

INSERT INTO doris_sink
SELECT user_id, TUMBLE_START(order_time, INTERVAL '1' MINUTE), SUM(amount)
FROM orders
GROUP BY user_id, TUMBLE(order_time, INTERVAL '1' MINUTE);

面试加分提示

💡 说清楚SQL的定位和实战经验：

「我们80%的实时任务用Flink SQL，只有复杂的自定义逻辑用DataStream API。SQL的优势是开发效率高，而且Planner会自动优化。

核心概念是动态表——流就是一张持续变化的表，查询持续运行产出结果。时间属性通过WATERMARK声明。

常用场景：实时聚合、TopN、去重、CDC同步。写入Doris直接用Flink Doris Connector。」

📌 总结

记忆点	内容
SQL定位	80%场景够用，开发效率远高于DataStream
动态表	流=不断变化的表，查询持续运行
时间属性	WATERMARK声明Event Time
高频场景	聚合/TopN/去重/CDC同步
优势	开发快+自动优化+易维护

▶ 下一篇预告：DAY 11 CDC与Flink

---

Wood聊大数据 | 9年大数据老兵，聊技术、聊职场、聊成长

---

小红书正文

我们80%的Flink任务都用SQL写。原来DataStream写半天的任务SQL半小时搞定。

核心概念：动态表。在Flink SQL里流就是一张不断变化的表，查询持续运行持续产出结果。

时间属性在建表时用WATERMARK声明。窗口聚合用TUMBLE/HOP/SESSION函数。

高频场景：① 实时聚合——GROUP BY + TUMBLE窗口 ② TopN——ROW_NUMBER + PARTITION BY ③ 去重——ROW_NUMBER取第一条 ④ 写Doris——用Flink Doris Connector直接INSERT INTO。

DataStream API只在需要自定义State逻辑、复杂窗口触发的场景才用。