流式 SQL：Flink SQL 与 ksqlDB

流（Stream）  = 无限的、按时间追加的事件序列
动态表（Dynamic Table） = 流在某时刻的物化视图，随新事件不断变化

stream → [持续查询] → 动态表 → 输出 changelog

CREATE TABLE user_click_count (
  user_id   BIGINT,
  cnt       BIGINT,
  PRIMARY KEY (user_id) NOT ENFORCED
) WITH (
  'connector' = 'upsert-kafka',
  'topic' = 'user_click_count',
  'key.format' = 'json',
  'value.format' = 'json'
);

CREATE TABLE clicks (
  user_id     BIGINT,
  page_id     STRING,
  click_time  TIMESTAMP(3),
  WATERMARK FOR click_time AS click_time - INTERVAL '5' SECOND
) WITH (
  'connector' = 'kafka',
  'topic' = 'clicks',
  'properties.bootstrap.servers' = 'kafka:9092',
  'properties.group.id' = 'clicks-job',
  'scan.startup.mode' = 'group-offsets',
  'format' = 'json',
  'json.ignore-parse-errors' = 'true'
);

-- 1. 处理时间（Processing Time）：用引擎当前时间，无需水位线
proc_time AS PROCTIME()

-- 2. 事件时间（Event Time）：从数据里取，需 WATERMARK 声明
WATERMARK FOR click_time AS click_time - INTERVAL '5' SECOND

-- 3. 事件时间列的两种写法
--   DDL 中声明（推荐，全表统一）
--   查询中用 WATERMARK 派生

-- 滚动窗口：每分钟每页面 PV
SELECT
  window_start, window_end, page_id, COUNT(*) AS pv
FROM TABLE(TUMBLE(TABLE clicks, DESCRIPTOR(click_time), INTERVAL '1' MINUTE))
GROUP BY window_start, window_end, page_id;

-- 滑动窗口：每 30 秒统计过去 5 分钟
SELECT window_start, window_end, COUNT(*) AS pv
FROM TABLE(HOP(TABLE clicks, DESCRIPTOR(click_time), INTERVAL '30' SECOND, INTERVAL '5' MINUTE))
GROUP BY window_start, window_end;

-- 累积窗口：适合"从当天 0 点到现在的累计"报表
SELECT window_start, window_end, COUNT(*) AS cumulative_pv
FROM TABLE(CUMULATE(TABLE clicks, DESCRIPTOR(click_time), INTERVAL '10' MINUTE, INTERVAL '1' DAY))
GROUP BY window_start, window_end;

-- 会话窗口
SELECT window_start, window_end, COUNT(*) AS session_pv
FROM TABLE(SESSION(TABLE clicks, DESCRIPTOR(click_time), INTERVAL '30' MINUTE))
GROUP BY window_start, window_end;

-- Interval Join：订单与 30 分钟内的支付关联
SELECT o.order_id, o.amount, p.pay_time
FROM orders o
JOIN payments p
  ON o.order_id = p.order_id
 AND p.pay_time BETWEEN o.order_time AND o.order_time + INTERVAL '30' MINUTE;

-- Lookup Join：实时查 MySQL 维表
SELECT o.order_id, c.city
FROM orders AS o
JOIN dim_customer FOR SYSTEM_TIME AS OF o.proc_time AS c
  ON o.user_id = c.user_id;

-- 开启状态 TTL，避免状态无限增长（Flink 1.18+ 表级配置）
SET 'table.exec.state.ttl' = '24h';

-- 微批聚合：用延迟换吞吐，减少对下游的写放大
SET 'table.exec.mini-batch.enabled' = 'true';
SET 'table.exec.mini-batch.allow-latency' = '5s';
SET 'table.exec.mini-batch.size' = '5000';

-- 开启两阶段聚合，缓解热点 Key
SET 'table.optimizer.agg-phase-strategy' = 'TWO_PHASE';

-- 从 topic 建流
CREATE STREAM clicks (
  user_id BIGINT,
  page_id VARCHAR,
  click_time TIMESTAMP
) WITH (
  KAFKA_TOPIC = 'clicks',
  VALUE_FORMAT = 'JSON',
  TIMESTAMP_FORMAT = 'yyyy-MM-dd HH:mm:ss'
);

-- 物化成表（会创建新 topic 并持续更新）
CREATE TABLE page_pv AS
SELECT page_id, COUNT(*) AS pv
FROM clicks
WINDOW TUMBLING (SIZE 1 MINUTE)
GROUP BY page_id
EMIT CHANGES;

-- Push Query：持续推送结果，永不结束（流式查询）
SELECT page_id, COUNT(*) FROM clicks EMIT CHANGES;

-- Pull Query：对物化表做点查，立即返回当前值（类似数据库查询）
SELECT * FROM page_pv WHERE page_id = 'home' LIMIT 1;

-- 1. 消费 Debezium 变更流
CREATE TABLE orders_cdc (
  order_id  BIGINT,
  user_id   BIGINT,
  amount    DECIMAL(12,2),
  status    STRING,
  ts        TIMESTAMP(3),
  WATERMARK FOR ts AS ts - INTERVAL '10' SECOND
) WITH (
  'connector' = 'kafka',
  'topic' = 'dbserver1.inventory.orders',
  'format' = 'debezium-json',
  'properties.bootstrap.servers' = 'kafka:9092'
);

-- 2. 每分钟 GMV（只统计已支付）
INSERT INTO gmv_per_minute
SELECT
  window_start, window_end,
  SUM(amount) AS gmv,
  COUNT(DISTINCT user_id) AS buyers
FROM TABLE(TUMBLE(TABLE orders_cdc, DESCRIPTOR(ts), INTERVAL '1' MINUTE))
WHERE status = 'paid'
GROUP BY window_start, window_end;

1. 语法与计划校验   提交前用 sql-gateway / TableEnvironment 编译一次，拦住语法错
2. 确定性输入输出测试 用 datagen 连接器或固定输入表，跑一遍断言 changelog
3. 结果 diff        影子作业双跑，对比输出 topic 的 changelog 是否一致

-- 测试用输入：固定 10 条，可复现
CREATE TABLE clicks_test (
  user_id BIGINT, page_id STRING, click_time TIMESTAMP(3),
  WATERMARK FOR click_time AS click_time - INTERVAL '5' SECOND
) WITH (
  'connector' = 'datagen',
  'number-of-rows' = '10',
  'fields.user_id.kind' = 'sequence',
  'fields.user_id.start' = '1',
  'fields.user_id.end' = '10'
);