4. Flink SQL 与特征计算：使用 Flink SQL 定义实时特征、CREATE TABLE AS SELECT、窗口聚合与特征物化

在实时特征服务中，Flink SQL 提供了声明式的特征定义能力，将流式数据处理抽象为标准的 SQL 操作。本章聚焦于如何利用 Flink SQL 完成实时特征的计算、窗口聚合以及特征结果的物化存储，帮助团队以较低的开发成本构建高性能的特征管道。

4.1 实时特征的定义方式

Flink SQL 通过 CREATE TABLE 语句将外部系统（Kafka、消息队列、数据库）映射为逻辑表，特征计算逻辑则通过 SELECT 查询表达。实时特征通常分为三类：

4.2 CREATE TABLE AS SELECT（CTAS）与特征物化

CREATE TABLE AS SELECT（CTAS）是 Flink SQL 中实现特征物化的核心语法。它将查询结果持续写入目标表，完成从“计算”到“存储”的闭环。CTAS 在特征服务中的典型用法如下：

-- 定义源表：用户行为事件流
CREATE TABLE user_behavior (
    user_id      STRING,
    event_type   STRING,
    event_time   TIMESTAMP(3),
    page_id      STRING,
    duration     BIGINT,
    WATERMARK FOR event_time AS event_time - INTERVAL '5' SECOND
) WITH (
    'connector' = 'kafka',
    'topic' = 'user_behavior',
    'properties.bootstrap.servers' = 'localhost:9092',
    'format' = 'json'
);

-- 定义结果表：物化特征存储
CREATE TABLE feature_store (
    user_id      STRING,
    window_start TIMESTAMP(3),
    window_end   TIMESTAMP(3),
    click_count  BIGINT,
    avg_duration DOUBLE,
    PRIMARY KEY (user_id, window_start) NOT ENFORCED
) WITH (
    'connector' = 'upsert-kafka',
    'topic' = 'feature_output',
    'properties.bootstrap.servers' = 'localhost:9092',
    'key.format' = 'json',
    'value.format' = 'json'
);

-- CTAS：实时计算并物化特征
INSERT INTO feature_store
SELECT
    user_id,
    TUMBLE_START(event_time, INTERVAL '5' MINUTE) AS window_start,
    TUMBLE_END(event_time, INTERVAL '5' MINUTE)   AS window_end,
    COUNT(*)                                       AS click_count,
    AVG(duration)                                  AS avg_duration
FROM user_behavior
WHERE event_type = 'click'
GROUP BY
    user_id,
    TUMBLE(event_time, INTERVAL '5' MINUTE);

4.3 窗口聚合与特征时效性

窗口聚合是实时特征计算中最常用的模式。Flink SQL 支持三种窗口类型，各自适用于不同的特征时效性要求：

窗口聚合特征的计算延迟直接影响特征服务的实时性。对于低延迟场景，建议使用 滑动窗口 配合较短的滑动步长（如 10 秒），以获取近似连续的特征更新。

-- 滑动窗口示例：每 10 秒计算近 5 分钟的平均响应时间
SELECT
    api_id,
    HOP_START(response_time, INTERVAL '10' SECOND, INTERVAL '5' MINUTE) AS window_start,
    HOP_END(response_time, INTERVAL '10' SECOND, INTERVAL '5' MINUTE)   AS window_end,
    AVG(latency_ms)                                                      AS avg_latency
FROM api_metrics
GROUP BY
    api_id,
    HOP(response_time, INTERVAL '10' SECOND, INTERVAL '5' MINUTE);

4.4 特征物化策略与存储选型

特征物化是指将实时计算的特征结果持久化到外部存储，供在线推理服务低延迟读取。不同的物化策略对应不同的存储选型：

在 Flink SQL 中，物化策略通过 WITH 参数和连接器配置实现。例如，使用 upsert-kafka 连接器实现增量物化，使用 jdbc 连接器配合 lookup 模式实现全量物化。

4.5 特征计算管道架构

下图展示了基于 Flink SQL 的实时特征计算管道整体架构，从数据源接入到特征物化输出的完整链路：

4.6 特征物化的关键配置与优化

在实际生产环境中，特征物化的性能取决于多个配置参数的调优。以下表格列出了 CTAS 物化场景下的关键配置项：

4.7 窗口聚合特征计算的完整示例

以下示例展示了一个完整的实时特征计算流程，包含事件时间窗口聚合、维度关联和特征物化：

-- 1. 定义行为事件源表
CREATE TABLE click_events (
    user_id   STRING,
    item_id   STRING,
    category  STRING,
    click_ts  TIMESTAMP(3),
    WATERMARK FOR click_ts AS click_ts - INTERVAL '10' SECOND
) WITH (
    'connector' = 'kafka',
    'topic' = 'click_events',
    'properties.bootstrap.servers' = 'kafka:9092',
    'format' = 'json'
);

-- 2. 定义维度表（支持 Lookup Join）
CREATE TABLE item_dim (
    item_id   STRING,
    item_name STRING,
    price     DOUBLE,
    PRIMARY KEY (item_id) NOT ENFORCED
) WITH (
    'connector' = 'jdbc',
    'url' = 'jdbc:mysql://mysql:3306/feature_db',
    'table-name' = 'item_dim',
    'lookup.cache.max-rows' = '10000',
    'lookup.cache.ttl' = '10min'
);

-- 3. 定义特征结果表
CREATE TABLE item_feature (
    item_id      STRING,
    window_start TIMESTAMP(3),
    window_end   TIMESTAMP(3),
    click_count  BIGINT,
    avg_price    DOUBLE,
    category     STRING,
    PRIMARY KEY (item_id, window_start) NOT ENFORCED
) WITH (
    'connector' = 'upsert-kafka',
    'topic' = 'item_feature',
    'properties.bootstrap.servers' = 'kafka:9092',
    'key.format' = 'json',
    'value.format' = 'json'
);

-- 4. CTAS 执行特征计算与物化
INSERT INTO item_feature
SELECT
    c.item_id,
    TUMBLE_START(c.click_ts, INTERVAL '1' MINUTE) AS window_start,
    TUMBLE_END(c.click_ts, INTERVAL '1' MINUTE)   AS window_end,
    COUNT(*)                                        AS click_count,
    AVG(d.price)                                    AS avg_price,
    MAX(d.category)                                 AS category
FROM click_events c
JOIN item_dim FOR SYSTEM_TIME AS OF c.click_ts d
    ON c.item_id = d.item_id
GROUP BY
    c.item_id,
    TUMBLE(c.click_ts, INTERVAL '1' MINUTE);

4.8 特征物化的监控与数据一致性

特征物化过程中需要关注数据一致性和管道健康度。以下为关键监控指标：

• 数据延迟： 监控 Watermark 与当前时间的差值，判断特征计算是否滞后。
• 写入 QPS： 监控目标存储的写入速率，避免背压导致物化延迟。
• 主键冲突率： Upsert 场景下监控主键冲突次数，排查特征重复或状态异常。
• 状态大小： 监控 Flink 算子状态大小，防止大状态导致 Checkpoint 超时。

对于数据一致性要求极高的场景（如金融风控特征），建议启用 端到端 Exactly-Once 语义，配合 Kafka 事务写入或两阶段提交协议，确保特征结果不重不丢。