RisingWave 技术文档（四）

应用篇：实战场景与案例分析

“理论的价值在于指导实践。这一篇我们将展示 RisingWave 如何在实际业务场景中大显身手。”

4.1 实时分析报表

场景描述

电商平台需要在双十一大促期间，实时展示各品类销售额、订单量、用户行为等核心指标。

技术方案

使用 RisingWave 构建实时数据管道：

-- 创建订单源表
CREATE TABLE orders (
    order_id BIGINT PRIMARY KEY,
    user_id BIGINT,
    category VARCHAR,
    amount DECIMAL,
    created_at TIMESTAMP
) WITH (
    connector = 'kafka',
    kafka.topic = 'orders',
    kafka.brokers = 'localhost:9092'
);
-- 创建实时物化视图：每小时销售额
CREATE MATERIALIZED VIEW hourly_sales AS
SELECT
    date_trunc('hour', created_at) AS hour,
    category,
    SUM(amount) AS total_amount,
    COUNT(*) AS order_count
FROM orders
GROUP BY date_trunc('hour', created_at), category;
-- 创建级联视图：实时 TOP 10 品类
CREATE MATERIALIZED VIEW top_categories AS
SELECT * FROM (
    SELECT
        category,
        SUM(amount) AS total,
        ROW_NUMBER() OVER (ORDER BY SUM(amount) DESC) AS rank
    FROM orders
    WHERE created_at > NOW() - INTERVAL '24' HOUR
    GROUP BY category
) WHERE rank <= 10;

4.2 监控报警系统

场景描述

金融交易系统需要实时监控异常交易，当某项指标超过阈值时立即报警。

技术方案

-- 交易监控表
CREATE TABLE transactions (
    tx_id BIGINT,
    user_id BIGINT,
    amount DECIMAL,
    risk_score DECIMAL,
    created_at TIMESTAMP
);
-- 高风险交易告警物化视图
CREATE MATERIALIZED VIEW high_risk_alerts AS
SELECT
    user_id,
    COUNT(*) AS alert_count,
    MAX(risk_score) AS max_risk,
    SUM(amount) AS total_amount
FROM transactions
WHERE risk_score > 0.8
  AND created_at > NOW() - INTERVAL '5' MINUTE
GROUP BY user_id
HAVING COUNT(*) > 5 OR SUM(amount) > 100000;
-- 窗口聚合：5分钟滚动统计
CREATE MATERIALIZED VIEW tx_stats_5min AS
SELECT
    window_start,
    window_end,
    COUNT(*) AS tx_count,
    AVG(amount) AS avg_amount,
    MAX(amount) AS max_amount,
    PERCENTILE_CONT(0.99) WITHIN GROUP (ORDER BY amount) AS p99_amount
FROM TUMBLING(
    transactions,
    INTERVAL '5' MINUTE,
    created_at
)
GROUP BY window_start, window_end;

4.3 流式 ETL

场景描述

从多个异构数据源（MySQL、PostgreSQL、外部 API）实时同步数据到数据仓库。

技术方案

-- MySQL 订单数据 CDC
CREATE SOURCE mysql_orders (...) WITH (connector = 'mysql_cdc', ...);
-- PostgreSQL 产品目录
CREATE SOURCE pg_products (...) WITH (connector = 'postgres_cdc', ...);
-- 外部 API 用户数据
CREATE SOURCE api_users (...) WITH (connector = 'kafka', ...);
-- 合并订单视图
CREATE MATERIALIZED VIEW enriched_orders AS
SELECT
    o.order_id,
    o.user_id,
    u.username,
    u.email,
    p.product_name,
    p.category,
    o.amount,
    o.created_at
FROM mysql_orders o
JOIN api_users u ON o.user_id = u.user_id
JOIN pg_products p ON o.product_id = p.product_id;
-- Sink 到 ClickHouse
CREATE SINK enriched_orders_sink AS
SELECT * FROM enriched_orders
WITH (
    connector = 'clickhouse',
    clickhouse.url = 'clickhouse://localhost:9000',
    clickhouse.database = 'analytics'
);

4.4 机器学习特征工程

场景描述

实时计算用户特征，用于推荐系统和风控模型。

技术方案

-- 用户行为事件流
CREATE SOURCE user_events (
    user_id BIGINT,
    event_type VARCHAR,
    product_id BIGINT,
    amount DECIMAL,
    created_at TIMESTAMP
);
-- 实时用户特征
CREATE MATERIALIZED VIEW user_features AS
SELECT
    user_id,
    -- 统计特征
    COUNT(*) AS event_count,
    SUM(amount) AS total_amount,
    AVG(amount) AS avg_amount,
    -- 行为特征
    COUNT(DISTINCT event_type) AS behavior_diversity,
    COUNT(DISTINCT product_id) AS product_diversity,
    -- 时间特征
    MAX(created_at) AS last_active,
    -- 窗口特征（最近1小时）
    SUM(CASE WHEN created_at > NOW() - INTERVAL '1' HOUR THEN 1 ELSE 0 END) AS hourly_events
FROM user_events
GROUP BY user_id;

结合 UDF Server 实现复杂特征计算：

-- 注册 Python UDF 进行特征标准化
CREATE FUNCTION normalize_features(features JSONB)
RETURNS JSONB
LANGUAGE python AS $$
import json
from sklearn.preprocessing import StandardScaler
def normalize_features(features):
    data = json.loads(features)
    values = [[d['event_count'], d['total_amount'], d['avg_amount']]]
    scaler = StandardScaler()
    normalized = scaler.fit_transform(values)[0]
    return json.dumps({
        'event_count_norm': normalized[0],
        'total_amount_norm': normalized[1],
        'avg_amount_norm': normalized[2]
    })
$$;

4.5 案例分析

案例一：某电商平台实时大屏

业务背景：
双十一期间需要实时展示全站销售额、订单量、UV 等核心指标。

传统架构：
Kafka → Flink → Redis → 大屏（延迟 30 秒+）

改造后架构：
MySQL CDC → RisingWave → 大屏（延迟 < 1 秒）

效果对比：

• 数据延迟：从 30 秒降至 < 1 秒
• 运维成本：降低 60%
• 系统稳定性：SLA 99.99%

案例二：某金融公司风控系统

业务背景：
实时检测异常交易，延迟要求毫秒级。

技术架构：
PostgreSQL → RisingWave → 规则引擎 → 告警

关键 SQL：

CREATE MATERIALIZED VIEW fraud_detection AS
SELECT
    user_id,
    card_id,
    SUM(amount) AS total,
    COUNT(*) AS tx_count,
    MAX(risk_score) AS max_risk
FROM transactions
WHERE created_at > NOW() - INTERVAL '10' MINUTE
GROUP BY user_id, card_id
HAVING SUM(amount) > 50000 OR COUNT(*) > 20 OR MAX(risk_score) > 0.9;

4.6 生态集成

复制到HTML复制到Markdown复制到Excel

4.7 小结

RisingWave 的典型应用场景：

1. 实时分析报表：电商大屏、运营监控
2. 监控报警：金融风控、异常检测
3. 流式 ETL：数据同步、格式转换
4. 特征工程：推荐系统、风控模型
5. 实时 BI：用户行为分析

下一篇预告

面对众多技术方案，如何做出正确选择？下一篇我们将进行横向对比：

• RisingWave vs Apache Flink
• RisingWave vs ClickHouse
• RisingWave vs 传统数据库
• RisingWave vs Materialize
• 选型决策树

敬请期待：《对比篇：选型指南与方案对比》

想了解更多？

• 官方文档：https://docs.risingwave.com/
• 中文教程：https://www.risingwavetutorial.com/
• GitHub：https://github.com/risingwavelabs/risingwave