3. SQL-on-Streaming：流式SQL引擎（Flink SQL / ksqlDB）的基本语法与窗口概念

在组合风险归因场景中，数据以事件流的形式持续到达。传统的批处理SQL无法处理“无界”数据，因此需要流式SQL引擎。本章聚焦于Flink SQL与ksqlDB的核心语法与窗口机制，帮助您将风险计算从离线迁移至实时。

3.1 流式SQL引擎对比：Flink SQL vs. ksqlDB

两种引擎均支持ANSI SQL子集与流式语义，但在架构定位与语法细节上存在差异。下表从风险归因视角进行对比：

3.2 流式SQL核心语法：从Source到Sink

流式SQL的声明式管道遵循“定义Source → 定义计算逻辑 → 定义Sink”的三段式结构。以下以Flink SQL为例展示完整流程：

-- 1. 定义Kafka Source（持仓变更事件流）
CREATE TABLE position_changes (
  portfolio_id   STRING,
  asset_id       STRING,
  quantity       DECIMAL(20,4),
  market_price   DECIMAL(20,4),
  event_time     TIMESTAMP(3) METADATA FROM ‘timestamp’,
  WATERMARK FOR event_time AS event_time - INTERVAL ‘5’ SECOND
) WITH (
  ‘connector’ = ‘kafka’,
  ‘topic’ = ‘position-events’,
  ‘properties.bootstrap.servers’ = ‘localhost:9092’,
  ‘format’ = ‘json’
);

-- 2. 定义Sink（风险归因结果表）
CREATE TABLE risk_attribution_sink (
  window_start    TIMESTAMP(3),
  portfolio_id    STRING,
  total_var       DECIMAL(30,6),
  asset_contrib   STRING   -- JSON格式存储各资产贡献
) WITH (
  ‘connector’ = ‘kafka’,
  ‘topic’ = ‘risk-attribution-results’,
  ‘format’ = ‘json’
);

-- 3. 流式查询：滚动窗口内计算VaR贡献
INSERT INTO risk_attribution_sink
SELECT
  TUMBLE_START(event_time, INTERVAL ‘1’ MINUTE) AS window_start,
  portfolio_id,
  SUM(market_price * quantity) AS total_var,
  LISTAGG(CONCAT(asset_id, ‘:’, CAST(market_price * quantity AS STRING)), ‘,’) AS asset_contrib
FROM position_changes
GROUP BY
  TUMBLE(event_time, INTERVAL ‘1’ MINUTE),
  portfolio_id;

3.3 窗口概念：时间驱动的风险聚合

窗口是流式SQL将无界数据切分为有界计算单元的核心机制。在组合风险归因中，窗口决定了“多久计算一次风险值”以及“哪些事件属于同一计算批次”。

以下SVG展示了三种基础窗口在风险归因流水线中的位置与数据流：

3.4 窗口语法详解与风险归因示例

以下分别给出Flink SQL与ksqlDB中三种窗口的语法模板，并附上组合风险归因的实战代码。

3.4.1 滚动窗口（Tumbling Window）

滚动窗口将数据流切分为固定大小、不重叠的时间段。适用于每分钟计算一次组合风险敞口。

-- Flink SQL
SELECT
  TUMBLE_START(event_time, INTERVAL ‘5’ MINUTE) AS window_start,
  portfolio_id,
  SUM(market_price * quantity) AS gross_exposure
FROM position_changes
GROUP BY TUMBLE(event_time, INTERVAL ‘5’ MINUTE), portfolio_id;

-- ksqlDB (5分钟滚动窗口)
SELECT
  WINDOWSTART AS window_start,
  portfolio_id,
  SUM(market_price * quantity) AS gross_exposure
FROM position_changes
WINDOW TUMBLING (SIZE 5 MINUTES)
GROUP BY portfolio_id
EMIT CHANGES;

3.4.2 滑动窗口（Hopping Window）

滑动窗口允许窗口重叠，适合需要高频刷新但计算粒度较粗的场景，如每30秒输出过去5分钟的归因结果。

-- Flink SQL (滑动窗口：每30秒滑动，窗口大小5分钟)
SELECT
  HOP_START(event_time, INTERVAL ‘30’ SECOND, INTERVAL ‘5’ MINUTE) AS window_start,
  HOP_END(event_time, INTERVAL ‘30’ SECOND, INTERVAL ‘5’ MINUTE) AS window_end,
  portfolio_id,
  AVG(market_price * quantity) AS avg_exposure
FROM position_changes
GROUP BY
  HOP(event_time, INTERVAL ‘30’ SECOND, INTERVAL ‘5’ MINUTE),
  portfolio_id;

-- ksqlDB (滑动窗口：每30秒跳跃，窗口大小5分钟)
SELECT
  WINDOWSTART AS window_start,
  WINDOWEND AS window_end,
  portfolio_id,
  AVG(market_price * quantity) AS avg_exposure
FROM position_changes
WINDOW HOPPING (SIZE 5 MINUTES, ADVANCE BY 30 SECONDS)
GROUP BY portfolio_id
EMIT CHANGES;

3.4.3 会话窗口（Session Window）

会话窗口根据事件之间的不活动间隔划分窗口，特别适合交易行为分析——当交易员连续操作时归入同一会话，间隔超过阈值则开启新会话。

-- Flink SQL (会话窗口：不活动间隔2分钟)
SELECT
  SESSION_START(event_time, INTERVAL ‘2’ MINUTE) AS session_start,
  SESSION_END(event_time, INTERVAL ‘2’ MINUTE) AS session_end,
  trader_id,
  SUM(notional) AS session_trading_volume
FROM trade_events
GROUP BY SESSION(event_time, INTERVAL ‘2’ MINUTE), trader_id;

-- ksqlDB (会话窗口：不活动间隔2分钟)
SELECT
  WINDOWSTART AS session_start,
  WINDOWEND AS session_end,
  trader_id,
  SUM(notional) AS session_trading_volume
FROM trade_events
WINDOW SESSION (2 MINUTES)
GROUP BY trader_id
EMIT CHANGES;

3.5 窗口进阶：累计窗口与迟到数据处理

在组合风险归因中，累计窗口（Cumulate Window）是Flink SQL独有的特性，它从窗口起始点开始累积数据，每次触发输出当前累积结果，直到窗口结束。这对于“从开盘到当前时刻的累计风险”场景非常有用。

-- Flink SQL 累计窗口：每1分钟输出从当天开盘到现在的累计VaR
SELECT
  portfolio_id,
  SUM(market_price * quantity) AS cumulative_var
FROM position_changes
GROUP BY
  CUMULATE(event_time, INTERVAL ‘1’ MINUTE, INTERVAL ‘1’ DAY),
  portfolio_id;

以下SVG展示了累计窗口与迟到数据处理的内部机制：

3.6 窗口选择决策清单

在组合风险归因中，选择合适的窗口类型直接影响计算准确性与延迟。以下清单帮助快速决策：

• 固定频率风险报告（如每分钟、每小时）→ 滚动窗口，语法简洁，结果可预测。
• 高频监控+低频计算（如每10秒查看过去5分钟风险）→ 滑动窗口，注意重叠导致的计算量放大。
• 交易行为驱动（如一次交易会话内的风险累积）→ 会话窗口，需合理设置不活动间隔。
• 从开盘到当前的累计风险（如盘中实时VaR）→ 累计窗口（仅Flink SQL），避免自维护状态。
• 数据乱序严重（如跨地域交易事件）→ 优先使用事件时间 + 设置合理的 WATERMARK 与 ALLOWED_LATENESS。