RisingWave 技术文档（六）

上手篇：5 分钟入门 RisingWave

“Talk is cheap, show me the code。这可能是最简单易懂的流数据库入门教程，让你在 5 分钟内体验从零构建实时数据 pipeline 的快感。”

6.1 安装与部署

RisingWave 支持多种部署方式，从单机的 Docker 体验到生产级的 Kubernetes 集群，总有一款适合你。

Docker 快速体验（5 分钟上手）

最快的方式是用 Docker 运行 RisingWave：

# 启动一个单节点 RisingWave 实例
docker run -it \
  --name risingwave \
  -p 4566:4566 \
  -p 5691:5691 \
  risingwavelabs/risingwave:latest \
  ./target/release/risingwave \
  --backend-state-store=mem

参数说明：

• 4566：PostgreSQL 兼容协议端口（连接客户端用）
• 5691：Meta 服务端口
• --backend-state-store=mem：使用内存存储（仅适合体验，不适合生产）

注意： 如果你已经有 Docker Compose 的 RisingWave 环境，可以跳过上面这步。

连接 RisingWave

RisingWave 兼容 PostgreSQL 协议，你可以用任何 PostgreSQL 客户端连接：

# 使用 psql 连接
psql -h localhost -p 4566 -U root

# 使用 Python 连接
import psycopg2
conn = psycopg2.connect(
    host="localhost",
    port=4566,
    user="root",
    database="dev"
)
cursor = conn.cursor()
cursor.execute("SELECT version();")
print(cursor.fetchone())

预期输出：

                                                     version
----------------------------------------------------------------------------------------------------------------
 PostgreSQL 16.2 (RisingWave 1.5.0-dev (e5e2d66fc on 2024-04-01T10:00:00Z) compiled by gcc (Apple clang 15.0.0))
(1 row)

恭喜你，RisingWave 已经成功运行！

单机调试模式

如果你是开发者，想要在本地调试 RisingWave 源码：

# 克隆源码
git clone https://github.com/risingwavelabs/risingwave.git
cd risingwave
# 使用 risedev 启动本地集群
./risedev d

./risedev d 会启动一个完整的开发集群（Frontend + Compute + Meta + MinIO），适合本地调试和功能验证。

Docker Compose 集群部署

生产环境推荐使用 Docker Compose：

# docker-compose.yml
version: '3'
services:
  meta:
    image: risingwavelabs/risingwave:latest
    command: meta-node
    ports:
      - "5690:5690"
    volumes:
      - meta_data:/risingwave/meta_node
  compute:
    image: risingwavelabs/risingwave:latest
    command: compute-node
    depends_on:
      - meta
    ports:
      - "5688:5688"
    environment:
      - META_ADDR=meta:5690
  frontend:
    image: risingwavelabs/risingwave:latest
    command: frontend-node
    depends_on:
      - meta
    ports:
      - "4566:4566"
    environment:
      - META_ADDR=meta:5690
  minio:
    image: minio/minio:latest
    command: server /data --console-address ":9001"
    ports:
      - "9000:9000"
      - "9001:9001"
volumes:
  meta_data:

启动集群：

docker-compose up -d

Kubernetes 部署

生产级部署推荐使用 Kubernetes，配合 Operator 实现自动化管理：

# 添加 Helm 仓库
helm repo add risingwave https://charts.risingwave.com
helm repo update
# 安装 RisingWave
helm install risingwave risingwave/risingwave \
  --set meta.nodeSelector.type=compute \
  --set compute.replicas=3 \
  --set storage.type=aws-s3

详细的 Kubernetes 配置请参考官方 K8s 部署文档。

6.2 第一个物化视图

让我们从零开始，构建你的第一个实时物化视图。

场景：实时统计每分钟订单金额

假设有一个订单表，实时接收订单数据，我们想要实时统计每分钟的订单总额。

步骤一：创建数据源

RisingWave 支持多种数据源。我们先创建一个模拟的 Kafka Source：

-- 创建一个 Kafka Source
CREATE SOURCE orders (
    order_id BIGINT,
    user_id BIGINT,
    amount DECIMAL,
    created_at TIMESTAMP
)
WITH (
    connector = 'kafka',
    kafka.topic = 'orders',
    kafka.brokers = 'localhost:9092',
    kafka.scan.startup.mode = 'earliest'
)
FORMAT PLAIN ENCODE JSON;

如果你没有真实的 Kafka，也可以用 Table Source 模拟：

-- 直接创建 Table，数据需要手动插入
CREATE TABLE orders (
    order_id BIGINT PRIMARY KEY,
    user_id BIGINT,
    amount DECIMAL,
    created_at TIMESTAMP
);

步骤二：插入测试数据

-- 向 orders 表插入测试数据
INSERT INTO orders VALUES
    (1, 1001, 99.50, NOW()),
    (2, 1002, 150.00, NOW()),
    (3, 1001, 299.00, NOW()),
    (4, 1003, 75.25, NOW());

步骤三：创建物化视图

现在创建第一个物化视图，实时统计每分钟订单金额：

CREATE MATERIALIZED VIEW order_stats_by_minute AS
SELECT
    date_trunc('minute', created_at) AS minute,
    COUNT(*) AS order_count,
    SUM(amount) AS total_amount,
    AVG(amount) AS avg_amount
FROM orders
GROUP BY date_trunc('minute', created_at);

步骤四：查询物化视图

-- 查询实时统计结果
SELECT * FROM order_stats_by_minute;

预期输出：

         minute         | order_count | total_amount | avg_amount
------------------------+-------------+---------------+-------------
 2026-04-01 10:00:00   |           4 |       623.75  |    155.9375
(1 row)

步骤五：观察增量更新

现在我们再插入一条数据，观察物化视图如何实时更新：

-- 插入新订单
INSERT INTO orders VALUES
    (5, 1004, 200.00, NOW());
-- 再次查询
SELECT * FROM order_stats_by_minute;

注意到 order_count 和 total_amount 已经自动更新了——这就是 RisingWave 增量计算的能力。

6.3 进阶操作

时间窗口

Tumbling Window（滚动窗口）

滚动窗口是固定大小、不重叠的窗口：

-- 每小时统计
CREATE MATERIALIZED VIEW hourly_orders AS
SELECT
    window_start,
    window_end,
    COUNT(*) AS order_count,
    SUM(amount) AS total_amount
FROM TUMBLING(
    orders,
    INTERVAL '1' HOUR,
    created_at
)
GROUP BY window_start, window_end;

Sliding Window（滑动窗口）

滑动窗口可以定义滑动步长：

-- 每 5 分钟统计一次最近 30 分钟的订单
CREATE MATERIALIZED VIEW sliding_orders AS
SELECT
    window_start,
    window_end,
    COUNT(*) AS order_count,
    SUM(amount) AS total_amount
FROM SLIDING(
    orders,
    INTERVAL '30' MINUTE,
    INTERVAL '5' MINUTE,
    created_at
)
GROUP BY window_start, window_end;

多流 Join

RisingWave 支持多个流的 Join，这是流处理中最强大的能力之一：

-- 场景：订单流 + 用户信息流 = 带用户详情的订单流
-- 创建用户信息 Source
CREATE SOURCE users (
    user_id BIGINT,
    username VARCHAR,
    email VARCHAR
)
WITH (
    connector = 'kafka',
    kafka.topic = 'users',
    kafka.brokers = 'localhost:9092'
)
FORMAT PLAIN ENCODE JSON;
-- 创建带用户信息的订单 MV
CREATE MATERIALIZED VIEW orders_with_users AS
SELECT
    o.order_id,
    o.amount,
    o.created_at,
    u.username,
    u.email
FROM orders o
JOIN users u ON o.user_id = u.user_id;

水位线（Watermark）

处理乱序数据时，水位线是关键机制：

-- 定义水位线，允许最多 5 分钟的乱序
CREATE SOURCE orders_with_watermark (
    order_id BIGINT,
    user_id BIGINT,
    amount DECIMAL,
    created_at TIMESTAMP,
    WATERMARK FOR created_at AS created_at - INTERVAL '5' MINUTE
)
WITH (
    connector = 'kafka',
    kafka.topic = 'orders',
    kafka.brokers = 'localhost:9092'
)
FORMAT PLAIN ENCODE JSON;

6.4 生产环境配置

资源规划

CPU 和内存配置

# risingwave.yml
# Compute 节点资源配置
compute:
  # 每个 Actor 的线程数
  actor_runtime_worker_threads: 4
# Meta 节点资源配置
meta:
  # Meta 服务线程数
  meta_worker_threads: 4
# 存储配置
storage:
  # 共享 buffer 缓存大小（建议为机器内存的 1/4）
  share_buffer_capacity_mb: 4096

存储容量规划

Hummock 使用对象存储（S3/MinIO），容量取决于：

• 基表数据量
• 物化视图数量和复杂度
• Checkpoint 保留策略
• Compaction 策略

估算公式：

存储容量 ≈ (基表数据量 × MV 数量 × 状态膨胀系数) + Checkpoint 缓冲区

通常状态膨胀系数为 2-5 倍。

监控与运维

关键监控指标

RisingWave 暴露了丰富的 Prometheus 指标：

# prometheus.yml
scrape_configs:
  - job_name: 'risingwave'
    static_configs:
      - targets: ['risingwave:5691']  # Meta 节点

关键指标：

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

日志管理

# risingwave.yml
logging:
  # 日志级别
  level: info
  # 日志输出格式
  format: json
  # 日志保留天数
  retention_days: 30

高可用配置

多副本配置

# risingwave.yml
compute:
  # 每个 Fragment 的副本数
  replicas: 3
meta:
  # Meta 节点高可用
  max_retries: 3

当副本数为 3 时，系统可以容忍 1 个节点故障而不中断服务。

故障转移

RisingWave 的故障转移是自动的：

1. Meta 节点检测到 Compute 节点心跳超时
2. 将该节点的任务调度到其他节点
3. 新节点从 Checkpoint 恢复状态
4. 物化视图「追上」最新状态

整个过程对用户透明，无需人工干预。

6.5 性能调优

查询优化

查看执行计划

-- 使用 EXPLAIN 查看查询计划
EXPLAIN (VERBOSE) SELECT * FROM order_stats_by_minute;

使用索引优化查询

-- 为高频查询列创建索引
CREATE INDEX idx_orders_user_id ON orders(user_id);
CREATE INDEX idx_orders_created_at ON orders(created_at);

状态管理

控制状态大小

对于窗口较大的查询，状态可能增长很快：

-- 设置窗口过期时间
CREATE MATERIALIZED VIEW recent_orders AS
SELECT *
FROM orders
WHERE created_at > NOW() - INTERVAL '24' HOUR;

Checkpoint 调优

# risingwave.yml
storage:
  # Checkpoint 间隔（默认 10 秒）
  checkpoint_interval_ms: 10000
  # Checkpoint 并行度
  checkpoint_concurrency: 8

建议：

• 低延迟场景：5 秒
• 标准场景：10 秒（默认）
• 大状态场景：30-60 秒

6.6 常见问题与解决方案

Q1: 物化视图更新延迟高

可能原因：

• 状态太大，Checkpoint 成为瓶颈
• Compute 节点资源不足

解决方案：

-- 查看当前延迟
SELECT * FROM rw_materialized_views;
-- 优化：增加 Checkpoint 并行度
-- 优化：增加 Compute 节点资源

Q2: Kafka Source 消费慢

可能原因：

• Kafka 分区数不足
• Consumer 并发数不够

解决方案：

-- 增加 Kafka 分区数（在 Kafka 端操作）
-- 创建 Source 时指定并发度
CREATE SOURCE orders (...)
WITH (
    connector = 'kafka',
    kafka.topic = 'orders',
    kafka.brokers = 'localhost:9092',
    kafka.consumer_threads = 8  -- 增加并发
)
FORMAT PLAIN ENCODE JSON;

Q3: 多流 Join 状态爆炸

可能原因：

• Join 的窗口太大
• Join 键的区分度低

解决方案：

• 使用 Tumbling Window 限制状态生命周期
• 选择区分度高的 Join 键

6.7 社区资源

官方文档

• 英文文档：https://docs.risingwave.com/
• 中文文档：https://docs.risingwave.com/docs/current/
• 官方教程：https://www.risingwavetutorial.com/

GitHub

• 主仓库：https://github.com/risingwavelabs/risingwave
• Issue 反馈：https://github.com/risingwavelabs/risingwave/issues
• Discussion：https://github.com/risingwavelabs/risingwave/discussions

社区交流

• Slack：https://risingwave-community.slack.com/
• 微信公众号：RisingWave
• 知乎：https://www.zhihu.com/people/risingwave

贡献指南

如果你对 RisingWave 的开发感兴趣：

# 克隆源码
git clone https://github.com/risingwavelabs/risingwave.git
# 查看贡献指南
cat CONTRIBUTING.md
# 搭建开发环境
./risedev d

快速参考

常用 SQL 命令

-- 查看所有表
\dt
-- 查看所有物化视图
\dv
-- 查看源
SHOW SOURCES;
-- 查看连接器
SHOW CONNECTORS;
-- 刷新物化视图（仅限静态 MV）
REFRESH MATERIALIZED VIEW mv_name;

6.8 小结

恭喜你完成了 RisingWave 入门之旅！让我们回顾今天学到的内容：

核心要点：

1. 安装部署：Docker 是最快的体验方式，Docker Compose 适合开发，Kubernetes 适合生产
2. 物化视图：一条 SQL 即可创建持续更新的实时视图
3. 时间窗口：Tumbling Window 和 Sliding Window 满足不同分析需求
4. 多流 Join：支持 10-20 个流的同时 Join
5. 水位线：优雅处理乱序和延迟数据
6. 生产配置：合理的资源规划和监控是高可用的基础

下一步建议：

1. 用 Docker 快速体验核心功能
2. 尝试连接真实的 Kafka 或 PostgreSQL CDC
3. 构建一个属于你的实时分析 Pipeline
4. 加入 Slack 社区，与开发者直接交流

恭喜你完成了 RisingWave 技术文档系列！

从流数据库的核心概念，到系统架构、核心特性、实战场景、选型对比，再到本篇的快速入门，你应该已经对 RisingWave 有了全面且深入的理解。

如果你还没有尝试过，现在就是最好的时机：

docker run -it --name risingwave -p 4566:4566 risingwavelabs/risingwave:latest ./target/release/risingwave --backend-state-store=mem

然后用 psql -h localhost -p 4566 -U root 连接，开始你的实时计算之旅吧！

想了解更多？

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