Posted inデータベース

Apache Pinotを極める：ミリ秒単位の分析を実現するリアルタイムOLAP

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

大規模リアルタイム分析の課題

秒間50,000イベントが発生するプラットフォーム向けのリアルタイムダッシュボード構築は、標準的なデータベースにとって悪夢です。ステークホルダーは、売上合計、売れ筋商品、エンゲージメント指標が即座に更新されることを期待します。多くのチームがMySQL、PostgreSQL、あるいはMongoDBをこの役割に無理やり当てはめようとするのを見てきましたが、通常はすぐに限界にぶつかります。

テーブルが1億行を超えると、リレーショナルデータベースにおける単純なGROUP BYやCOUNT(DISTINCT)クエリは極端に遅くなります。アグレッシブにインデックスを貼ったとしても、データ量が増えるにつれてパフォーマンスの低下は避けられません。サマリーテーブルにデータを事前集計するのは一般的な回避策ですが、ユーザーが特定のGPS座標、デバイスモデル、カスタム時間範囲などの任意のディメンションでフィルタリングする必要がある場合、この戦術は通用しません。

根本原因：なぜ伝統的なデータベースはOLAPに不向きなのか

ボトルネックの原因がハードウェアのパワー不足であることは稀です。問題の本質はアーキテクチャにあります。PostgreSQLなどのデータベースは、オンライン取引処理（OLTP）向けに構築されています。これらはデータを「行」単位で保存します。これは単一顧客の残高を更新するには最適ですが、平均価格を計算するために5億行をスキャンするには非効率的です。

Apache Pinotは、分散型の列指向OLAPデータストアとして、このギャップを埋めます。Star-tree、Bloomフィルタ、Rangeインデックスなどの特殊なインデックス作成機能を使用し、ペタバイト規模のデータセットに対してもミリ秒単位のクエリレイテンシを維持します。PrestoやTrinoのようなエンジンがデータのある場所（S3など）でクエリを実行するのに対し、Pinotはデータを独自の高度に最適化された形式で保存します。そのため、ミリ秒が重要となるユーザー向けの分析機能において最適な選択肢となります。

代替案の比較

Pinotを採用する前に、いくつかの競合製品を評価しました。それぞれに得意分野がありますが、ライブデータのストリーム処理能力には大きな違いがあります。

機能 Apache Pinot ClickHouse Presto / Trino
主なユースケース ユーザー向けリアルタイムアプリ 社内BIおよびログ分析 複数ソースを跨ぐフェデレーションSQL
取り込みレイテンシ 1秒未満（真のリアルタイム） 数秒（マイクロバッチ） 高い（ソースに依存）
クエリレイテンシ ミリ秒単位 (p99 < 200ms) ミリ秒〜数秒 数秒〜数分
ストレージ戦略 高度な列指向インデックス 列指向MergeTree 分離型 (S3, HDFS)

Apache Pinotのトレードオフ

強み

  • 驚異的なスピード: UIのレスポンスを維持するために、200ミリ秒未満で返さなければならないクエリ専用に構築されています。
  • ネイティブなKafka統合: PinotはKafkaやKinesisを第一級市民として扱うため、リアルタイムの取り込みがシームレスに行えます。
  • 水平方向の弾力性: クエリの同時実行数やストレージ容量の増加に合わせて、ノードを追加することでクラスタを拡張できます。
  • Star-Treeインデックス: この独自の機能により、Pinotはデータを事前集計しつつ、ユーザーがローデータまでドリルダウンすることを可能にします。

弱み

  • 運用の複雑さ: クラスタの管理には、Controller、Broker、Server、およびZookeeperインスタンスの調整が必要です。
  • イミュータブル（不変）なデータ: Pinotは追記型のストリーム向けに設計されています。特定の行に対してUPDATEやDELETEを実行することは、伝統的な意味ではサポートされていません。
  • 厳格なスキーマ: データ型を変更する必要がある場合、多くの場合、データセグメントの再取り込みが必要になります。

実践的な開発セットアップ

本番環境は通常Kubernetesで動作しますが、ローカルでの統合テストにはDocker Composeが最も速い方法です。機能的なPinotクラスタは、いくつかの動的なパーツで構成されます。Controllerはクラスタの状態を管理し、Brokerはクエリをルーティングします。Serverはデータの保存とスキャンの実行という重労働を担います。最後に、Zookeeperがこれらすべてのコンポーネントの同期を保ちます。

Apache Kafkaとのステップバイステップ統合

KafkaからPinotにイベントを直接ストリーミングし、即座にクエリを実行できるパイプラインを構築してみましょう。

1. インフラの起動

このdocker-compose.ymlを使用して、PinotとKafkaを一緒に起動します。この設定は、リアルタイムテスト用のサンドボックスを提供します。

version: '3.7'
services:
  zookeeper:
    image: zookeeper:3.8
    ports:
      - "2181:2181"
  pinot-controller:
    image: apachepinot/pinot:latest
    command: "StartController -zkAddress zookeeper:2181"
    ports:
      - "9000:9000"
    depends_on:
      - zookeeper
  pinot-broker:
    image: apachepinot/pinot:latest
    command: "StartBroker -zkAddress zookeeper:2181"
    ports:
      - "8099:8099"
    depends_on:
      - pinot-controller
  pinot-server:
    image: apachepinot/pinot:latest
    command: "StartServer -zkAddress zookeeper:2181"
    depends_on:
      - pinot-broker
  kafka:
    image: bitnami/kafka:latest
    environment:
      - KAFKA_CFG_ZOOKEEPER_CONNECT=zookeeper:2181
      - ALLOW_PLAINTEXT_LISTENER=yes
    ports:
      - "9092:9092"
    depends_on:
      - zookeeper

2. データストリームの生成

orders-topicという名前のトピックを作成します。ライブの購入イベントをシミュレートするために、JSONメッセージを投入します。

# トピックを作成
docker exec -it kafka /opt/bitnami/kafka/bin/kafka-topics.sh --create --topic orders-topic --bootstrap-server localhost:9092

# サンプル注文データを投入
echo '{"order_id": 101, "product_name": "メカニカルキーボード", "price": 150.00, "timestamp": 1672531200000}' | \
docker exec -i kafka /opt/bitnami/kafka/bin/kafka-console-producer.sh --topic orders-topic --bootstrap-server localhost:9092

3. スキーマの定義

Pinotがデータ型を理解するにはスキーマが必要です。これをorders_schema.jsonとして保存します。フィルタリング用のディメンションと計算用のメトリクスを区別します。

{
  "schemaName": "orders",
  "dimensionFieldSpecs": [
    {"name": "order_id", "dataType": "LONG"},
    {"name": "product_name", "dataType": "STRING"}
  ],
  "metricFieldSpecs": [
    {"name": "price", "dataType": "DOUBLE"}
  ],
  "dateTimeFieldSpecs": [{
    "name": "timestamp",
    "dataType": "LONG",
    "format": "1:MILLISECONDS:EPOCH",
    "granularity": "1:MILLISECONDS"
  }]
}

4. リアルタイムテーブルの設定

テーブル設定により、PinotにKafkaストリームの場所を教えます。これをorders_table.jsonとして保存します。Kafkaブローカーを指し示すstreamConfigsセクションに注目してください。

{
  "tableName": "orders",
  "tableType": "REALTIME",
  "segmentsConfig": {
    "timeColumnName": "timestamp",
    "schemaName": "orders",
    "replication": "1"
  },
  "tableIndexConfig": {
    "loadMode": "MMAP",
    "streamConfigs": {
      "streamType": "kafka",
      "stream.kafka.consumer.type": "lowlevel",
      "stream.kafka.topic.name": "orders-topic",
      "stream.kafka.decoder.class.name": "org.apache.pinot.plugin.inputformat.json.JSONMessageDecoder",
      "stream.kafka.consumer.prop.auto.offset.reset": "smallest",
      "stream.kafka.broker.list": "kafka:9092"
    }
  }
}

5. 設定の適用

これらのファイルをPinot Controllerに送信して、取り込みプロセスを開始します。

docker exec -it pinot-controller /opt/pinot/bin/pinot-admin.sh AddTable \
  -schemaFile /path/to/orders_schema.json \
  -tableConfigFile /path/to/orders_table.json \
  -exec

パフォーマンスのテスト

http://localhost:9000でPinotクエリコンソールを開きます。ライブストリームに対して標準的なSQLを実行できるようになります。製品ごとの収益を計算してみましょう。

SELECT 
    product_name, 
    COUNT(*), 
    SUM(price) 
FROM orders 
GROUP BY product_name 
ORDER BY SUM(price) DESC

結果は、Kafkaに新しいメッセージを投入したほぼ瞬間に更新されます。私の本番環境のデプロイでは、ユーザーが「購入」をクリックしてからPinotにデータが表示されるまでのラグは、一貫して500ミリ秒未満です。

結論

伝統的なRDBMSからApache Pinotへの移行は、データ処理方法における大きな転換です。しかし、成長スピードがデータベースの集計能力を追い越してしまった場合、それは必要なステップです。分析の重労働をPinotにオフロードすることで、トランザクションデータベースを軽量に保つことができます。不正検知、ライブリーダーボード、またはリアルタイム監視を伴うプロジェクトであれば、Pinotはスタックに不可欠なツールとなるでしょう。

Share:
Tags: