Posted inDatabase

Apache Pinot：リアルタイムOLAPデータベースのインストールガイド — 100ms以下のアナリティクスダッシュボード構築

Database tutorial - IT technology blog
Database tutorial - IT technology blog

Apache Pinotが本当に必要なのはどんな時か？

インストールの話をする前に、実際の体験談を一つ紹介したい。昨年、私はあるECプラットフォームのアナリティクスシステムを担当していた。最初はMySQL単体を使っていて、「日別売上」ダッシュボードのクエリは3〜5秒で問題なかった。ところがデータが5000万行を超えた頃、その数字が30〜40秒になってしまった。ユーザーからのクレームは絶えず、PMからはデッドライン通告が来て、早急に解決策を見つける必要があった。

データベースの移行も検討した。以前、MySQLからPostgreSQLへの100GBのマイグレーションを経験したことがあり、計画に3日、実行に1日かかった。ただ今回はPostgreSQLも解決策にはならなかった。問題の本質はOLAP（オンライン分析処理）であり、OLTPではないからだ。そこで出会ったのがApache Pinotだった。

Apache Pinotは、LinkedInが開発し2015年にオープンソース化した分散OLAPデータストアだ。LinkedIn、Uber、Walmartがユーザー向けアナリティクス——エンドユーザーが直接参照するダッシュボードで1秒以上の待機が許されない場面——に利用している。

なぜPinotは速いのか？アーキテクチャが根本的に異なるからだ：

  • カラム型ストレージ：行全体ではなく必要な列だけを読み込むため、I/Oを大幅に削減
  • Star-treeインデックス：ディメンションごとにデータを事前集計し、GROUP BYクエリを10〜100倍高速化
  • リアルタイム＋バッチインジェスト：KafkaやCSV/JSON/Parquetファイルからデータを取り込み可能
  • 水平スケーリング：ダウンタイムなしでサーバーを追加可能

10億行のデータに対して100ms以下でクエリ結果を返すダッシュボード？それこそまさにPinotが解くために設計された問題だ。

Apache Pinotのインストール

Dockerでインストール（開発・テスト環境に推奨）

ローカルの開発・テスト環境では、Dockerの方がはるかに手軽だ——Javaを別途インストールする必要がなく、classpathの設定も不要。DockerとDocker Composeがすでに用意されている前提で、docker-compose.ymlファイルを作成する：

version: '3.7'
services:
  zookeeper:
    image: zookeeper:3.5.6
    hostname: zookeeper
    ports:
      - "2181:2181"
    environment:
      ZOOKEEPER_CLIENT_PORT: 2181
      ZOOKEEPER_TICK_TIME: 2000

  pinot-controller:
    image: apachepinot/pinot:1.2.0
    command: "StartController -zkAddress zookeeper:2181"
    restart: unless-stopped
    ports:
      - "9000:9000"
    depends_on:
      - zookeeper

  pinot-broker:
    image: apachepinot/pinot:1.2.0
    command: "StartBroker -zkAddress zookeeper:2181"
    restart: unless-stopped
    ports:
      - "8099:8099"
    depends_on:
      - pinot-controller

  pinot-server:
    image: apachepinot/pinot:1.2.0
    command: "StartServer -zkAddress zookeeper:2181"
    restart: unless-stopped
    ports:
      - "8098:8098"
    depends_on:
      - pinot-controller
docker-compose up -d

# コンテナの起動状態を確認
docker-compose ps

Controllerが完全に起動するまで30〜60秒ほど待つ。その後ブラウザでhttp://localhost:9000を開くと、Pinot UIが表示される。

バイナリでインストール（本番環境）

# Pinot 1.2.0 をダウンロード
wget https://downloads.apache.org/pinot/apache-pinot-1.2.0/apache-pinot-1.2.0-bin.tar.gz
tar -xvf apache-pinot-1.2.0-bin.tar.gz
cd apache-pinot-1.2.0-bin

# ZooKeeperを起動
bin/pinot-admin.sh StartZookeeper -zkPort 2181 &

# Controllerを起動
bin/pinot-admin.sh StartController \
  -zkAddress localhost:2181 \
  -controllerPort 9000 &

# Brokerを起動
bin/pinot-admin.sh StartBroker -zkAddress localhost:2181 &

# Serverを起動
bin/pinot-admin.sh StartServer -zkAddress localhost:2181 &

詳細設定

Pinotを動かすには3つのものが必要だ：Schema（データ構造）、Table Config（ストレージとインデックスの設定）、そしてData Ingestion（データの取り込み方法）。以下ではECサイトの注文データセットを使って説明する——イメージしやすく、実際のユースケースに近い内容だ。

ステップ1：スキーマの作成

スキーマは3種類のカラムでデータ構造を定義する：

  • dimensionFieldSpecs：フィルタリング/グルーピングに使用（category、user_idなど）
  • metricFieldSpecs：集計に使用（revenue、countなど）
  • dateTimeFieldSpecs：時刻カラム——必須項目

orders_schema.jsonファイルを作成する：

{
  "schemaName": "orders",
  "dimensionFieldSpecs": [
    {"name": "order_id", "dataType": "STRING"},
    {"name": "user_id", "dataType": "STRING"},
    {"name": "product_category", "dataType": "STRING"},
    {"name": "status", "dataType": "STRING"},
    {"name": "payment_method", "dataType": "STRING"}
  ],
  "metricFieldSpecs": [
    {"name": "revenue", "dataType": "DOUBLE"},
    {"name": "quantity", "dataType": "INT"},
    {"name": "discount", "dataType": "DOUBLE"}
  ],
  "dateTimeFieldSpecs": [
    {
      "name": "order_timestamp",
      "dataType": "TIMESTAMP",
      "format": "1:MILLISECONDS:EPOCH",
      "granularity": "1:MILLISECONDS"
    }
  ]
}
curl -X POST \
  "http://localhost:9000/schemas" \
  -H "Content-Type: application/json" \
  -d @orders_schema.json

ステップ2：インデックス付きテーブル設定の作成

インデックス設定こそ、Pinotがリレーショナルデータベースと根本的に異なる点だ。同じデータ、同じクエリでも——インデックスが適切なら50ms、不適切なら5秒になる。orders_table.jsonファイルを作成する：

{
  "tableName": "orders",
  "tableType": "OFFLINE",
  "segmentsConfig": {
    "timeColumnName": "order_timestamp",
    "timeType": "MILLISECONDS",
    "replication": "1",
    "schemaName": "orders"
  },
  "tableIndexConfig": {
    "loadMode": "MMAP",
    "invertedIndexColumns": ["product_category", "status", "payment_method"],
    "starTreeIndexConfigs": [
      {
        "dimensionsSplitOrder": ["product_category", "status"],
        "skipStarNodeCreationForDimensions": [],
        "functionColumnPairs": ["SUM__revenue", "COUNT__*"],
        "maxLeafRecords": 10000
      }
    ],
    "rangeIndexColumns": ["revenue", "order_timestamp"]
  },
  "tenants": {
    "broker": "DefaultTenant",
    "server": "DefaultTenant"
  },
  "metadata": {}
}

最もよく使われる3種類のインデックスで、それぞれ異なるクエリパターンに対応する：

  • invertedIndexColumns：完全一致の検索が非常に高速——WHERE status = 'COMPLETED'はO(1)で実行
  • starTreeIndexConfigs：ディメンションごとに事前集計を行い、GROUP BYクエリを通常のスキャンと比べて10〜100倍高速化
  • rangeIndexColumns：範囲クエリを高速化——WHERE revenue BETWEEN 500000 AND 2000000
curl -X POST \
  "http://localhost:9000/tables" \
  -H "Content-Type: application/json" \
  -d @orders_table.json

ステップ3：データの取り込み（バッチインジェスト）

サンプルCSVファイルsample_orders.csvを作成する：

order_id,user_id,product_category,status,payment_method,revenue,quantity,discount,order_timestamp
ORD001,USER100,Electronics,COMPLETED,CREDIT_CARD,1500000,1,0,1720310400000
ORD002,USER101,Fashion,COMPLETED,BANK_TRANSFER,350000,2,50000,1720310460000
ORD003,USER102,Electronics,PENDING,MOMO,2200000,1,100000,1720310520000
ORD004,USER103,Food,COMPLETED,CASH,85000,3,0,1720310580000

インジェストジョブの仕様ファイルingestion_job.yamlを作成する：

executionFrameworkSpec:
  name: 'standalone'
  segmentGenerationJobRunnerClassName: 'org.apache.pinot.plugin.ingestion.batch.standalone.SegmentGenerationJobRunner'
  segmentTarPushJobRunnerClassName: 'org.apache.pinot.plugin.ingestion.batch.standalone.SegmentTarPushJobRunner'

jobType: SegmentCreationAndTarPush

inputDirURI: '/tmp/pinot-data/input'
outputDirURI: '/tmp/pinot-data/output'
overwriteOutput: true

pinotFSSpecs:
  - scheme: file
    className: org.apache.pinot.spi.filesystem.LocalPinotFS

recordReaderSpec:
  dataFormat: 'csv'
  className: 'org.apache.pinot.plugin.inputformat.csv.CSVRecordReader'
  configClassName: 'org.apache.pinot.plugin.inputformat.csv.CSVRecordReaderConfig'

tableSpec:
  tableName: 'orders'
  schemaURI: 'http://localhost:9000/schemas/orders'
  tableConfigURI: 'http://localhost:9000/tables/orders'

pinotClusterSpecs:
  - controllerURI: 'http://localhost:9000'
# CSVを正しいディレクトリにコピー
mkdir -p /tmp/pinot-data/input
cp sample_orders.csv /tmp/pinot-data/input/

# インジェストを実行
bin/pinot-admin.sh LaunchDataIngestionJob \
  -jobSpecFile ingestion_job.yaml

動作確認とモニタリング

SQLクエリのテスト実行

PinotはSQL標準に対応している。http://localhost:9000でPinot UIを開き、Query Consoleタブで以下を試してみよう：

-- カテゴリ別売上
SELECT
  product_category,
  SUM(revenue)   AS total_revenue,
  COUNT(*)       AS order_count,
  AVG(revenue)   AS avg_order_value
FROM orders
WHERE status = 'COMPLETED'
GROUP BY product_category
ORDER BY total_revenue DESC
LIMIT 10

REST API経由でクエリする場合：

curl -X POST \
  "http://localhost:8099/query/sql" \
  -H "Content-Type: application/json" \
  -d '{
    "sql": "SELECT product_category, SUM(revenue) as total FROM orders GROUP BY product_category LIMIT 10"
  }'

レスポンスにはtimeUsedMsフィールドが含まれており、これが実際のクエリ時間だ。Star-treeインデックスが正しく設定されていれば、数百万行のデータでもこの値は通常50ms以下になる。

Pythonによるモニタリング

import requests

def query_pinot(sql: str, broker_url: str = "http://localhost:8099") -> dict:
    response = requests.post(
        f"{broker_url}/query/sql",
        headers={"Content-Type": "application/json"},
        json={"sql": sql},
        timeout=10
    )
    result = response.json()

    time_ms = result.get("timeUsedMs", 0)
    rows = result.get("resultTable", {}).get("rows", [])
    print(f"Query time: {time_ms}ms | Rows returned: {len(rows)}")

    return result

# 実際に計測する
result = query_pinot("""
    SELECT
      status,
      COUNT(*) AS cnt,
      SUM(revenue) AS total_revenue
    FROM orders
    GROUP BY status
    ORDER BY cnt DESC
""")

# 結果を表示
columns = result["resultTable"]["dataSchema"]["columnNames"]
for row in result["resultTable"]["rows"]:
    print(dict(zip(columns, row)))

クラスターヘルスの確認

# ヘルスチェック — クラスターが正常なら "OK" を返す
curl http://localhost:9000/health

# 作成済みテーブルの一覧
curl http://localhost:9000/tables

# ordersテーブルのセグメント情報
curl http://localhost:9000/segments/orders

# ブローカーのメトリクス（Prometheusフォーマット）
curl http://localhost:8099/metrics | grep pinot_broker

エラーが発生した場合は、ログを確認してデバッグする：

# Docker
docker logs pinot-controller --tail 50
docker logs pinot-server --tail 50

# バイナリ
tail -f logs/pinot-controller.log
tail -f logs/pinot-server.log

初心者が最もよく陥るミスは、star-treeインデックスの設定を忘れることだ。なくてもPinotは動くが、データ量が増えるとクエリ時間が急増する。期待より遅いと感じたら、tableIndexConfigを見直し、functionColumnPairsに適切なdimension/metricのペアを追加しよう。

MySQLとPostgreSQLはOLTPが得意——トランザクション、行単位のupdate・deleteなど。PinotはOLAPが得意——大規模データへの高速読み取りで、書き込みスループットは重視しない。問題が異なれば、使うツールも異なる。今回の設定で、自分のデータを実際に試してみてほしい——同じ5000万行でも、MySQLとPinotの差は30秒から1秒以下になることが多い。

Share:
Tags: