FastAPI × TimescaleDBでリアルタイム分析基盤を自作する！エンタープライズ級パイプラインの全貌

「自社サービスにアナリティクス機能を追加したいけど、SegmentやRudderStackは高すぎる…」そんな悩みを抱えたことはありませんか？

そのジレンマを解決するオープンソースのリアルタイム分析パイプラインがv1.0.0としてリリースされ、海外の開発者コミュニティで注目を集めています。今回はその仕組みと、実際にどう使えるかを日本語でわかりやすく解説します 🚀

Contents

🎯 これは何を解決するプロジェクト？
🏗️ アーキテクチャの全体像
💻 FastAPIでイベントを受け取るシンプルな例
🕐 TimescaleDBが時系列データに強い理由
まとめ

🎯 これは何を解決するプロジェクト？

data pipeline analytics / Photo by Negative Space via Pexels

SaaSプロダクトを運営していると、ほぼ必ずぶつかる問題があります。イベントトラッキング・リアルタイムダッシュボード・時系列集計という3点セットです。

選択肢はざっくり2つ。

✅ Segment / RudderStack などのSaaSに課金する
✅ 自前で構築する（でもゼロからは大変…）

このプロジェクトは「自前で構築」を、本番環境でも使えるクオリティでまるごとテンプレートとして提供してくれるものです。つまり、「ちゃんとした作り方」のお手本がオープンソースになったイメージです。

🏗️ アーキテクチャの全体像

このパイプラインは主に以下の技術スタックで構成されています。

⚡ FastAPI：高速なイベント受信エンドポイント
🐘 TimescaleDB：時系列データに特化したPostgreSQL拡張
📨 Redis / Kafka：非同期処理のキュー
📊 Grafana：リアルタイムダッシュボード

イメージとしては「玄関（FastAPI）→ 荷捌き場（Redis/Kafka）→ 倉庫（TimescaleDB）→ 見える化（Grafana）」という流れです。データが来たら即座に受け取り、順番に処理して、時系列でスパッと検索できる構造になっています。

💻 FastAPIでイベントを受け取るシンプルな例

核心部分のコードをシンプルに書くとこんな感じです。ポイントを先にまとめると：

Pydanticでイベントの型を厳密に定義
受信後は即レスポンスを返し、重い処理は非同期で流す
TimescaleDBへの書き込みはバッチ処理でまとめて効率化

# FastAPIでイベントを受け取る最小構成例
from fastapi import FastAPI
from pydantic import BaseModel
from datetime import datetime
import asyncpg

app = FastAPI()

# イベントのデータ構造を定義
class AnalyticsEvent(BaseModel):
    user_id: str
    event_name: str          # 例: "page_view", "button_click"
    properties: dict = {}    # 追加情報をJSONで自由に持てる
    timestamp: datetime = None

@app.post("/events")
async def track_event(event: AnalyticsEvent):
    # ここでRedisキューに積んで即レスポンス（ノンブロッキング）
    await enqueue_event(event)
    return {"status": "accepted"}  # 202 Acceptedイメージ

async def enqueue_event(event: AnalyticsEvent):
    # 実際はRedis/Kafkaに積む処理が入る
    # バックグラウンドワーカーがTimescaleDBへ書き込む
    pass

重要なのは「受信と保存を分離している」点です。イベントが爆発的に増えても、受信側はキューに積むだけなので詰まりません。

🕐 TimescaleDBが時系列データに強い理由

通常のPostgreSQLでも時系列データは扱えます。でもTimescaleDBを使うと、時間でデータを自動的に分割（ハイパーテーブル）してくれるため、数百万件のログでも高速に検索できます。

-- TimescaleDBのハイパーテーブル作成例
CREATE TABLE analytics_events (
    time        TIMESTAMPTZ NOT NULL,
    user_id     TEXT,
    event_name  TEXT,
    properties  JSONB
);

-- これだけで時系列最適化が有効になる！
SELECT create_hypertable('analytics_events', 'time');

たったこれだけの追加で、時間範囲でのクエリが劇的に速くなるんですよね。普通のDBとの違いが体感できるはずです。