ウェルネス分析のスケーリング：CQRSパターン実践ガイド

CQRSで読み書きを分離し、データ集約型アプリのボトルネックを解決。PostgreSQL、Kafka、ClickHouseでスケーラブルな分析システムを構築する方法を解説。

2025年12月11日読了時間 9分

このページの内容

健康とウェルネスのアプリケーションの世界では、データが王様です。ユーザーはワークアウトを記録し、食事を追跡し、睡眠をモニタリングすることで、常にデータの流れを生成します。逆に、何週間、何ヶ月、さらには何年にもわたる進捗を示す、美しく洞察に満ちたダッシュボードを表示することも期待しています。これは古典的なアーキテクチャ上の課題を生みます：データベースは、頻繁で単純な書き込み（例：「10kmランを記録」）と複雑で分析的な読み取り（例：「過去1年間の曜日別平均ランシングペースのトレンドを表示」）の両方に最適化されていなければなりません。

単一のデータベースが両方のタスクを処理する伝統的なモノリシックアーキテクチャでは、不可避免りにボトルネックにぶつかるでしょう。素早い書き込みに最適なデータモデル（正規化）は、快速な分析クエリには往々にして適さず、複雑な結合や集計が必要になることがあります。ここで**CQRS（コマンドクエリ責任分離）**パターンが活躍します。

CQRSは、データの読み取りと書き込みのモデルを分離するアーキテクチャパターンです。1つのモデルですべてを処理するのではなく、2つのモデルを用意します：

状態変更（書き込み、更新、削除）を処理するためのコマンドモデル
データを取得するためのクエリモデル

このチュートリアルでは、CQRSの力を実証する簡略化されたウェルネス分析システムを構築します。PostgreSQLデータベースで「書き込み」サイドを、ClickHouseデータベースで非正規化された「読み取り」サイドを作成し、すべてをKafkaイベントバスで接続します。このアーキテクチャにより、システムの各部分を独立してスケールさせることができ、素早いワークアウト記録と超高速な分析ダッシュボードの両方を確保します。✨

問題点の理解

典型的なウェルネスアプリを想像してください。書き込みオペレーションはシンプルで頻繁です：

LogWorkout (userId, type: 'run', duration: 3600, distance: 10)
AddMeal (userId, name: 'Salad', calories: 450)

これらは、データの整合性を確保するために、PostgreSQLのようなトランザクション的で正規化されたデータベースが最も適切に処理します。

しかし、読み取りオペレーションは別の問題です。ユーザーが次のようなクエリを含むダッシュボードを見たいとします：

「過去1年間の月別合計ラン距離は？」
「カロリー摂取量の週平均を表示してください」
「平日のワークアウト時間と週末のワークアウト時間を比較してください」

これらのクエリを正規化されたトランザクションデータベースに対して実行すると、遅くリソース集約的になり、複数の結合やすぐさまの計算が必要になることがあります。ユーザー基盤とデータ量が増えるにつれて、これらのクエリはアプリケーションを止めてしまいます。

これらのconcernsを分離することで、CQRSは各仕事に最適なツールを使用することを可能にします。トランザクション整合性に最適化された書き込みモデルと、分析ダッシュボードが実行する特定のクエリに合わせて非正規化され調整された読み取りモデルを持つことができます。

CQRSアーキテクチャパターン

次の図は、CQRSベースの分析システムを通じてデータが流れる様子を示しています：

この分離により、トランザクションデータベースを直接クエリする compared to 100倍高速な分析クエリが可能になります。

前提条件

開始する前に、以下についての基本的な理解が必要です：

REST API（例ではNode.jsとExpressを使用します）
DockerとDocker Compose（インフラストラクチャの実行用）
基本的なSQL

以下のインストールが必要です：

Node.js（v18以降）
Docker

PostgreSQLとKafkaで書き込みサイドを構築する

書き込みサイド、または「コマンド」サイドは、新しいワークアウトを記録するためのシンプルなAPIと、それらを保存するためのPostgreSQLデータベースで構成されます。

で行うこと

PostgreSQLでworkoutsテーブル用のシンプルで正規化されたスキーマを定義し、新しいレコードを挿入するためのAPIエンドポイントを作成します。

実装

まず、Docker Composeファイルを作成して、インフラストラクチャを実行しましょう：PostgreSQL、Kafka、ClickHouse。

yaml

# docker-compose.yml
version: '3.8'
services:
  postgres:
    image: postgres:15
    restart: always
    environment:
      - POSTGRES_USER=user
      - POSTGRES_PASSWORD=password
      - POSTGRES_DB=wellness_write_db
    ports:
      - '5432:5432'
    volumes:
      - postgres_data:/var/lib/postgresql/data

  zookeeper:
    image: confluentinc/cp-zookeeper:7.3.0
    environment:
      ZOOKEEPER_CLIENT_PORT: 2181

  kafka:
    image: confluentinc/cp-kafka:7.3.0
    depends_on: [zookeeper]
    ports:
      - "9092:9092"
    environment:
      KAFKA_BROKER_ID: 1
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:9092,PLAINTEXT_HOST://localhost:9092
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
      KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1

  clickhouse:
    image: clickhouse/clickhouse-server:latest
    ports:
      - "8123:8123"
      - "9000:9000"
    volumes:
      - clickhouse_data:/var/lib/clickhouse

volumes:
  postgres_data:
  clickhouse_data:

docker-compose up -dを実行して、すべてのサービスを開始します。

次に、コマンドAPI用のシンプルなNode.jsアプリケーションを設定しましょう。

bash

mkdir wellness-app
cd wellness-app
npm init -y
npm install express pg kafkajs

「書き込み」サービスのコードは次のとおりです。

javascript

// command-service/index.js
const express = require('express');
const { Pool } = require('pg');
const { Kafka } = require('kafkajs');

const app = express();
app.use(express.json());

// PostgreSQL setup
const pool = new Pool({
  user: 'user',
  host: 'localhost',
  database: 'wellness_write_db',
  password: 'password',
  port: 5432,
});

// Kafka setup
const kafka = new Kafka({
  clientId: 'wellness-app',
  brokers: ['localhost:9092']
});
const producer = kafka.producer();

// Create table if it doesn't exist
const initDb = async () => {
  await pool.query(`
    CREATE TABLE IF NOT EXISTS workouts (
      id SERIAL PRIMARY KEY,
      user_id VARCHAR(50) NOT NULL,
      type VARCHAR(50) NOT NULL,
      duration_seconds INT NOT NULL,
      distance_km FLOAT,
      logged_at TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP
    );
  `);
};

app.post('/log-workout', async (req, res) => {
  try {
    const { userId, type, durationSeconds, distanceKm } = req.body;
    // 1. Write to the transactional database
    const result = await pool.query(
      'INSERT INTO workouts(user_id, type, duration_seconds, distance_km) VALUES($1, $2, $3, $4) RETURNING *',
      [userId, type, durationSeconds, distanceKm]
    );
    const newWorkout = result.rows;
    console.log('Workout logged to PostgreSQL:', newWorkout);

    // 2. Publish an event to Kafka
    await producer.send({
      topic: 'workout-logged',
      messages: [{ value: JSON.stringify(newWorkout) }],
    });
    console.log('Event published to Kafka topic: workout-logged');

    res.status(201).json(newWorkout);
  } catch (error) {
    console.error('Failed to log workout:', error);
    res.status(500).send('Internal Server Error');
  }
});

const start = async () => {
  await initDb();
  await producer.connect();
  app.listen(3000, () => {
    console.log('Command service listening on port 3000');
  });
};

start();

仕組み

APIエンドポイント：/log-workoutという単一のエンドポイントがあり、新しいワークアウトレコードを受け入れます。
トランザクション書き込み：データはまずPostgreSQLのworkoutsテーブルに書き込まれます。これが私たちの「信頼できる情報源」です。オペレーションはアトミックであり、データの整合性を確保します。
イベントパブリッシング：データベースへの書き込みが成功すると、workout-loggedという名前のKafkaトピックにイベントを送信します。このイベントには、新しく作成されたワークアウトの完全な詳細が含まれています。これは、書き込みサイドと読み取りサイドを分離するための重要なステップです。

ClickHouseマテリアライズドビューで読み取りサイドを構築する

ここからは楽しい部分です：「読み取り」サイドです。私たちの目標は、分析ダッシュボードが必要とする正確なクエリに最適化されたClickHouse内のデータモデルを作成することです。

で行うこと

高速な分析クエリ用に設計されたClickHouse内にテーブルを作成します。ClickHouseは列指向データベースであり、OLAP（オンライン分析処理）ワークロードに対して極めて高速です。また、KafkaトピックからこのClickHouseテーブルにデータを取り込むメカニズムも設定します。

実装

ClickHouseには、Kafkaトピックに直接サブスクライブできるネイティブKafkaエンジンがあります。これにより、統合が非常にシームレスになります。

まず、ClickHouseインスタンスに接続します。Docker経由でコマンドラインクライアントを使用できます：

bash

docker-compose exec -it clickhouse clickhouse-client

ClickHouseクライアント内で、次のSQLコマンドを実行します：

sql

-- 1. 非正規化された分析データ用のターゲットテーブルを作成します。
-- これは、高速クエリ用にデータが永久に保存される場所です。
CREATE TABLE IF NOT EXISTS wellness_analytics.workouts_view
(
    `id` UInt32,
    `user_id` String,
    `type` String,
    `duration_seconds` UInt32,
    `distance_km` Float32,
    `logged_at` DateTime,
    `day_of_week` String,
    `is_weekend` UInt8
)
ENGINE = MergeTree()
PARTITION BY toYYYYMM(logged_at)
ORDER BY (user_id, logged_at);

-- 2. Kafkaエンジンテーブルを作成します。
-- このテーブルは、Kafkaトピックからのライブストリームとして機能します。データ自体は保存しません。
CREATE TABLE IF NOT EXISTS wellness_analytics.workouts_queue
(
    `id` UInt32,
    `user_id` String,
    `type` String,
    `duration_seconds` UInt32,
    `distance_km` Float32,
    `logged_at` DateTime
)
ENGINE = Kafka()
SETTINGS
    kafka_broker_list = 'kafka:9092',
    kafka_topic_list = 'workout-logged',
    kafka_group_name = 'clickhouse_wellness_consumer',
    kafka_format = 'JSONEachRow';

-- 3. マテリアライズドビューを作成します。
-- これは、Kafkaストリームをターゲット分析テーブルに接続する魔法です。
-- workouts_queue内の新しいメッセージを自動的にリッスンし、変換しながら
-- workouts_viewに挿入します。
CREATE MATERIALIZED VIEW IF NOT EXISTS wellness_analytics.workouts_consumer TO wellness_analytics.workouts_view
AS SELECT
    id,
    user_id,
    type,
    duration_seconds,
    distance_km,
    logged_at,
    formatDateTime(logged_at, '%A') AS day_of_week,
    if(in(toDayOfWeek(logged_at),), 1, 0) AS is_weekend
FROM wellness_analytics.workouts_queue;

仕組み

workouts_viewテーブル：これがメインの分析テーブルです。ClickHouseの分析用パワードライブであるMergeTreeエンジンを使用しました。day_of_weekやis_weekendなどの事前に計算された非正規化列が追加されていることに注目してください。これにより、クエリ時にこれらの計算を行う必要がなくなり、ダッシュボードが高速になります。
workouts_queueテーブル：このテーブルはKafkaエンジンを使用してKafkaブローカーに接続し、workout-loggedトピックをサブスクライブします。本質的に、トピック内のメッセージのライブストリーミングビューを提供します。
workouts_consumerマテリアライズドビュー：これがETL（抽出、変換、負荷）パイプラインです。workouts_queueから新しいレコードを引き出し、簡単な変換（day_of_weekの計算など）を適用し、workouts_viewテーブルにロードします。

最適化された読み取りモデルから分析をクエリする

書き込みサイドと読み取りサイドを設定したら、フロー全体をテストしましょう。

新しいワークアウトを投稿する

curlやPostmanなどのツールを使用して、コマンドサービスにPOSTリクエストを送信します：

bash

curl -X POST http://localhost:3000/log-workout \
-H "Content-Type: application/json" \
-d '{
    "userId": "user123",
    "type": "run",
    "durationSeconds": 3600,
    "distanceKm": 10.5
}'

ワークアウトがPostgreSQLに記録され、イベントがKafkaに送信されたことを示すコマンドサービスのコンソールにログが表示されるはずです。

ClickHouse内のデータを検証する

次に、ClickHouse読み取りモデルをクエリしましょう。ClickHouseクライアントに戻ります：

sql

SELECT * FROM wellness_analytics.workouts_view;

まもなく、day_of_weekとis_weekendフィールドが追加された新しいワークアウトレコードが表示されます！

複雑な分析クエリを実行する

これがアーキテクチャの真価を発揮する場所です。トランザクションデータベースでは遅かったクエリを実行してみましょう。例えば、「ユーザーごとのワークアウトタイプ別に合計距離と平均時間は？」

sql

SELECT
    user_id,
    type,
    sum(distance_km) as total_distance,
    avg(duration_seconds) / 60 as avg_duration_minutes
FROM wellness_analytics.workouts_view
GROUP BY user_id, type;

このクエリは、数百万または数十億の行があっても、ClickHouseでは信じられないほど高速です。これは、データがそのような集計に最適化された列指向形式で保存されているからです。

パフォーマンスに関する考慮事項

Independent Scaling：最大のボーナスは、readsとwritesを独立してスケールさせる能力です。分析ダッシュボードが人気になった場合、ワークアウト記録サービスの性能に影響を与えることなく、より多くのClickHouseノードを追加できます。
ClickHouse Optimization：ClickHouseのMergeTreeエンジンは高度に調整可能です（月별로パーティション分割したように）。データのことはパフォーマンスにとって重要で、ClickHouseが関連のないデータの読み取りをスキップできます。
結果整合性：このアーキテクチャは結果整合的であることを理解することが重要です。ワークアウトが記録されてから分析ダッシュボードに表示されるまで、わずかな遅延（通常はミリ秒から秒）があります。ほとんどのウェルネス分析ユースケースにとって、大規模な性能向上のトレードオフとしては完全に許容可能です。

代替アプローチ

Apache Druid：さらに低いレイテンシでリアルタイム取り込みとダッシュボードを必要とするユースケースには、ClickHouseの強力な代替手段としてApache Druidがあります。Druidのアーキテクチャはより分散しており、管理が複雑になる可能性がありますが、リアルタイム、イベント駆動のシナリオに優れています。
Change Data Capture（CDC）：アプリケーションがイベントを送信する代わりに、DebeziumなどのCDCツールを使用して、変更をPostgreSQLの先行書き込みログ（WAL）からKafkaに直接ストリーミングできます。これにより、書き込みサービスとイベントシステムがさらに分離されます。

結論

CQRSパターンを使用して、スケーラブルな分析バックエンドを正常に構築しました。書き込みモデル（PostgreSQL）を読み取りモデル（ClickHouse）から分離し、イベントバス（Kafka）で接続することで、次のようなシステムを作成しました：

高性能：トランザクション書き込みと分析読み取りの両方に最適化されたデータベース。
スケーラブル：需要を満たすためにreadsとwritesのサイドを独立してスケールできます。
保守性：concernsの分離により、システムの理解、開発、保守が容易になります。

パフォーマンスへの影響：健康分析システムでのCQRS実装は、複雑な集計で50〜100倍のクエリ性能向上を示しています。 организацииはダッシュボードの負荷時間90%削減と、同じインフラストラクチャで10倍の同時ユーザーを処理する能力を報告しています。ClickHouseの列指向ストレージにより、従来のSQLデータベースで数分かかる10億行のデータセットでの1秒未満の応答が可能になります。

このパターンは、电子商务プラットフォームからIoTシステム、そしてもちろんウェルネス分析まで、読み取りと書き込みのワークロードが一致していないあらゆるアプリケーション非常に強力です。

次のステップ

Query APIを構築する：ClickHouseのworkouts_viewをクエリするエンドポイントを提供し、フロントエンドダッシュボードにデータを提供する新しいサービスを作成します。
より豊かな分析を検討する：ClickHouseの強力なSQL方言と関数を使用して、時系列分析やファネル分析などのより複雑な分析を構築します。
エラー処理：ClickHouse消費者によって処理されなかったイベントを処理するために、Kafka内のデッドレターキューを実装します。

参考資料

Martin FowlerによるCQRSの公式ドキュメント
Kafkaエンジン用のClickHouseドキュメント
Apache Kafka公式サイト
PostgreSQL公式サイト
関連記事：
- Kafka＆Flinkによるリアルタイムパイプライン - スケールでのストリーム処理
- DDDでモノリスをマイクロサービスにリファクタリング - システムアーキテクチャパターン

クイックリファレンス

側面	重要な情報
トピック	Development
関連性	患者教育と認識
アクション	医療提供者と相談する
頻度	ケアチームのおすすめに応じて
エビデンス	現在の臨床ガイドラインに基づく

よくある質問

Q: これが自分に該当するどうかはどうすればわかりますか？ A: 個人的な病歴と家族歴について医療提供者と相談し、個々のリスク因子を評価し、適切な次のステップを決定してください。

Q: 医師に何を聞くべきですか？ A: 良い質問には：「結果は何を意味しますか？」「オプションは何がありますか？」「どんな生活習慣の改善が役立ちますか？」「いつフォローアップすべきですか？」があります。

Q: <delete_file>を助ける生活習慣の変更はありますか？ A: ほとんどの健康状態は、基本的な対策から恩恵を受けます：バランスの取れた栄養、定期的な身体活動（週150分以上）、質の高い睡眠（7〜9時間）、ストレス管理、そしてタバコの回避。

免責事項

この記事で提示されているアルゴリズムとテクニックは、純粋に技術教育の目的のみです。臨床的検証を受けておらず、医学的診断や治療決定に使用するべきではありません。医学的なアドバイスについては常に資格のある医療専門家にご相談ください。

すべての記事に戻る