RisingWave入門：PostgreSQL互換のストリーム処理データベースを使ってみる【2025年版】

はじめに：ストリーム処理とは？ RisingWaveが解決する課題

IoTデバイスからのセンサーデータ、金融取引、Webサイトのクリックログなど、現代のアプリケーションは絶え間なくデータを生成し続けています。これらの「データの流れ（ストリーム）」をリアルタイムで処理し、即座にインサイトを得る技術がストリーム処理です。

従来のストリーム処理は、Apache FlinkやApache Spark Streamingのような複雑なフレームワークを必要とし、専門的な知識が求められました。しかし、この分野に新しい波をもたらしているのが、RisingWaveです。

RisingWaveは、クラウドネイティブに設計された分散型SQLストリーム処理データベースです。その最大の特徴は、PostgreSQLと高い互換性を持つSQLを使って、高度なストリーム処理を記述できる点にあります。

RisingWaveが解決する課題:

• 複雑性の削減: Java/Scalaで複雑なコードを書く代わりに、使い慣れたSQLでストリーム処理を定義できます。
• リアルタイム性の実現: データが到着するたびにクエリ結果をインクリメンタルに更新する「マテリアライズドビュー」により、常に最新の分析結果を低レイテンシーで取得できます。
• エコシステムの活用: PostgreSQL互換であるため、psqlや各種BIツール、クライアントライブラリなど、既存のPostgreSQLエコシステムをそのまま活用できます。

本記事では、Dockerを使ってRisingWaveをローカルで動かし、簡単なストリーム処理を体験するまでの手順を解説します。

RisingWaveのアーキテクチャ概要

RisingWaveは、いくつかのコンポーネントから構成される分散システムです。

• Frontend: SQLクエリのパース、プランニング、最適化を行い、クライアントとの接続（psqlなど）を受け付けます。
• Compute Node: 実際のストリーム処理（データのフィルタリング、結合、集計など）を実行します。
• Meta Node: システム全体のメタデータ（テーブルスキーマ、ノードの状態など）を管理します。
• Object Storage: 状態（マテリアライズドビューの中間結果など）を永続化するために、S3互換のオブジェクトストレージを利用します。

このアーキテクチャにより、各コンポーネントを独立してスケールさせることが可能です。

ステップ1：Docker ComposeによるRisingWaveの起動

RisingWaveは、すべてのコンポーネントを含んだ便利なDocker Composeファイルを提供しています。これを使うと、ローカル環境で簡単にRisingWaveクラスタを起動できます。

# Docker Composeファイルをダウンロード
wget https://github.com/risingwavelabs/risingwave/releases/latest/download/risingwave-compose.yml

# 起動
docker-compose -f risingwave-compose.yml up -d

このコマンドで、Frontend, Compute, Metaの各ノードと、データソースのシミュレータであるdatagen、オブジェクトストレージのminioなどが起動します。

ステップ2：psqlを使った接続とデータソースの定義

RisingWaveはPostgreSQLのプロトコルを話すため、標準的なpsqlクライアントで接続できます。

psql -h localhost -p 4566 -d dev -U root

接続できたら、まず**データソース（Source）**を定義します。これは、KafkaやRedpanda、あるいは単純なTCPストリームなど、外部のデータストリームへの接続情報です。

ここでは、datagenが生成する架空のECサイトの注文データをソースとして定義してみましょう。

CREATE SOURCE orders (
    order_id INT,
    user_id INT,
    product_id INT,
    quantity INT,
    price DECIMAL
) WITH (
    connector = 'datagen',
    datagen.fields.order_id.kind = 'sequence',
    datagen.fields.user_id.kind = 'random',
    datagen.fields.user_id.min = '1',
    datagen.fields.user_id.max = '100',
    datagen.fields.product_id.kind = 'random',
    datagen.fields.product_id.min = '1',
    datagen.fields.product_id.max = '1000',
    datagen.fields.quantity.kind = 'random',
    datagen.fields.quantity.min = '1',
    datagen.fields.quantity.max = '5',
    datagen.fields.price.kind = 'random',
    datagen.fields.price.min = '10',
    datagen.fields.price.max = '1000',
    datagen.rows.per.second = '10'
) FORMAT PLAIN ENCODE JSON;

CREATE SOURCE文は、テーブル定義に似ていますが、データが永続化されるわけではなく、あくまで「流れてくるデータのスキーマ」を定義するものです。

ステップ3：マテリアライズドビューの作成とリアルタイム集計

RisingWaveの核心機能が**マテリアライズドビュー（Materialized View）**です。これは、クエリの結果を物理的に保存し、ソースデータが更新されるたびに自動的かつインクリメンタルに結果を更新し続けるオブジェクトです。

例えば、「商品ごとの売上高ランキング」をリアルタイムで集計するマテリアライズドビューを作成してみましょう。

CREATE MATERIALIZED VIEW product_sales_ranking AS
SELECT
    product_id,
    SUM(quantity * price) AS total_sales
FROM
    orders
GROUP BY
    product_id
ORDER BY
    total_sales DESC;

このSQLは、ごく普通の集計クエリです。しかし、CREATE MATERIALIZED VIEWとして定義することで、RisingWaveはこのクエリを継続的に実行するストリーム処理ジョブに変換します。ordersソースに新しいデータが流れ込むたびに、product_sales_rankingの内容が自動で更新されます。

ステップ4：データの流れ込みとリアルタイム更新の確認

マテリアライズドビューは、通常のテーブルと同じようにSELECT文でクエリできます。

-- 5秒待ってから実行
SELECT * FROM product_sales_ranking LIMIT 10;

 product_id | total_sales 
------------+-------------
        789 |     14567.89
        123 |     12345.67
        456 |      9876.54
... (結果は実行ごとに異なります)

さらに5秒待ってから同じクエリを再度実行してみてください。

-- 再度実行
SELECT * FROM product_sales_ranking LIMIT 10;

 product_id | total_sales 
------------+-------------
        789 |     18901.23  <-- 更新されている！
        123 |     15678.90  <-- 更新されている！
        456 |     11234.56
... 

total_salesの値が変化し、ランキングの順位も入れ替わっているかもしれません。これは、裏側でデータが流れ込み続け、マテリアライズドビューがリアルタイムで更新されている証拠です。バッチ処理のように定期的に集計を再実行する必要はありません。

まとめ：SQLでシンプルに始めるリアルタイムデータ分析

本記事では、RisingWaveを使って、PostgreSQL互換のSQLだけでリアルタイムのストリーム処理を実現する基本的な流れを体験しました。

• CREATE SOURCE でデータストリームの入り口を定義する。
• CREATE MATERIALIZED VIEW で集計や結合などの処理を定義し、結果を自動更新させる。
• あとは通常のデータベースのように**SELECT**するだけで、常に最新の分析結果を得られる。

RisingWaveは、これまで専門的なスキルセットが必要だったストリーム処理の世界を、多くのSQLに慣れ親しんだ開発者やデータアナリストに開放します。

リアルタイムのダッシュボード作成、異常検知、オンライン機械学習の特徴量生成など、その応用範囲は非常に広いです。複雑な分散システムを意識することなく、SQLで宣言的にストリーム処理を記述できるRisingWaveは、今後のリアルタイムデータ基盤において重要な選択肢となるでしょう。