PINOT

Last reviewed: 2026-09-17

Purpose: Apache Pinot real-time OLAP — architecture, schema and table configuration, stream and batch ingestion, SQL querying, and a worked streaming-analytics pipeline with Kafka and dashboards.

---

Table of Contents

• Overview
• Architecture
• Docker Compose Setup
• Schema Definition
• Table Configuration
• Ingestion: Streaming and Batch
• Transform Functions
• Querying with SQL
• Worked Example: Live Event Analytics
• Dashboards over Pinot
• Operations
• Topics Covered

---

Overview

Apache Pinot is a distributed, column-oriented OLAP datastore designed to answer analytical queries over event streams with sub-second latency at high concurrency. It ingests continuously from streaming platforms (Kafka, Kinesis, Pulsar) or in batches from object storage and Hadoop-style file systems, and serves SQL over the result — which makes it a good fit for user-facing analytics, real-time monitoring panels and dashboards where a warehouse's query latency would be too slow.

Typical position in a stack: producers → Kafka topic → Pinot (real-time table) → SQL clients, dashboards and APIs.

---

Architecture

Component	Role
Controller	Cluster coordination: manages tables, schemas, segments and assignment; exposes the admin UI
Broker	Receives queries, prunes segments, scatters to servers and merges results
Server	Stores segments and executes the per-segment scans and aggregations
Minion	Runs background tasks — segment compaction, roll-ups, purges, ingestion jobs
ZooKeeper	Stores cluster metadata and coordinates state (KRaft-style setups are available in newer releases)

Data is stored in segments, each covering a time range, and each segment carries its own indexes and metadata. A well-designed segment layout — partitioned by a high-cardinality column and sorted on the time or filter column most queries use — is worth more to query latency than any hardware upgrade.

---

Docker Compose Setup

A minimal single-node cluster is enough for development: ZooKeeper, controller, broker and server.

# compose.yaml — single-node Pinot for development
services:
  zookeeper:
    image: zookeeper:3.9
    ports: ["2181:2181"]

  pinot-controller:
    image: apachepinot/pinot:1.0.0
    command: StartController -zkAddress zookeeper:2181
    ports: ["9000:9000"]
    depends_on: [zookeeper]

  pinot-broker:
    image: apachepinot/pinot:1.0.0
    command: StartBroker -zkAddress zookeeper:2181
    ports: ["8099:8099"]
    depends_on: [pinot-controller]

  pinot-server:
    image: apachepinot/pinot:1.0.0
    command: StartServer -zkAddress zookeeper:2181
    ports: ["8098:8098"]
    depends_on: [pinot-controller]

docker compose up -d
docker compose logs -f pinot-controller
# Controller UI and REST API on :9000, broker query endpoint on :8099
curl -s http://localhost:9000/health | jq .

Run the servers with a JVM heap sized to the data volume (the Docker images accept -Xmx style arguments), and give Pinot its own ZooKeeper ensemble or a dedicated namespace in production — sharing an ensemble with other systems invites coordination failures.

---

Schema Definition

A schema declares the columns, their types, and which are dimensions, metrics and the time column. Send it to the controller once; tables reference it by name, and the schema enforces the type contract on ingest.

{
  "schemaName": "events",
  "primaryKeyColumns": ["event_id"],
  "dimensionFieldSpecs": [
    { "name": "event_id", "dataType": "STRING" },
    { "name": "user",     "dataType": "STRING" },
    { "name": "feature",  "dataType": "STRING" },
    { "name": "host",     "dataType": "STRING" },
    { "name": "is_test",  "dataType": "BOOLEAN" }
  ],
  "metricFieldSpecs": [
    { "name": "lic_count",    "dataType": "FLOAT" },
    { "name": "duration_secs", "dataType": "FLOAT" }
  ],
  "dateTimeFieldSpecs": [
    {
      "name": "event_time",
      "dataType": "TIMESTAMP",
      "format": "1:MILLISECONDS:EPOCH",
      "granularity": "1:MILLISECONDS"
    }
  ]
}

Notes worth getting right early:

• The dateTime field must be declared with the correct incoming format (EPOCH, SIMPLE_DATE_FORMAT, or a transform that produces one). Getting the format wrong is the most common reason an indexed table returns no rows for a time-range filter.
• Only declare as metrics the columns that will actually be aggregated; dimension columns are dictionary-encoded and expensive to aggregate over at scale.
• Use primaryKeyColumns on upsert-enabled tables to enforce one row per key.
• Nullability: missing fields become defaults, so validate upstream if a null and a zero mean different things.

---

Table Configuration

A table config selects REALTIME or OFFLINE, binds the schema, defines the segment layout, and — for streaming tables — declares the Kafka consumer settings.

{
  "tableName": "events",
  "tableType": "REALTIME",
  "segmentsConfig": {
    "timeColumnName": "event_time",
    "replication": "1",
    "retentionTimeUnit": "DAYS",
    "retentionTimeValue": "30"
  },
  "tableIndexConfig": {
    "loadMode": "MMAP",
    "streamConfigs": {
      "streamType": "kafka",
      "stream.kafka.consumer.type": "lowlevel",
      "stream.kafka.topic.name": "wiki-events",
      "stream.kafka.broker.list": "kafka:9093",
      "stream.kafka.decoder.class.name": "org.apache.pinot.plugin.stream.kafka.KafkaJSONMessageDecoder",
      "stream.kafka.consumer.factory.class.name": "org.apache.pinot.plugin.stream.kafka20.KafkaConsumerFactory",
      "realtime.segment.flush.threshold.rows": "1000",
      "realtime.segment.flush.threshold.time": "6h"
    }
  },
  "tenants": { "broker": "DefaultTenant", "server": "DefaultTenant" },
  "tableTypeDetails": {}
}

Setting	Effect
tableType	REALTIME consumes from a stream; OFFLINE loads pre-built segments from a file system
timeColumnName	Enables time-range pruning and retention; must be a schema dateTime column
retentionTime*	Age at which segments are dropped — the main lever on storage cost
replication	Segment copies per server; 2+ for production availability
loadMode	MMAP suits large tables and low memory overhead
flush.threshold.*	How much a consuming segment accumulates before it is sealed and indexed
aggregateMetrics	Pre-aggregates rows in real-time segments, cutting storage and query cost

Both an OFFLINE and a REALTIME table with the same name can be registered; Pinot queries them together as a hybrid table, which is how a batch backfill and a live stream are served from one logical table.

---

Ingestion: Streaming and Batch

Streaming (REALTIME). Pinot consumes the source topic directly; a low-level consumer gives each partition to one server, preserving ordering per partition. Events are written to a consuming segment, which is sealed — and a new one started — when the row, time or size threshold trips. Ingestion lag is visible in the controller UI and in pinot.server.realtime.* metrics.

Batch (OFFLINE). Pre-build segments from Parquet, Avro, JSON or CSV and push them to the controller with a batch ingestion job:

docker exec pinot-controller bin/pinot-admin.sh LaunchDataIngestionJob \
  -jobSpecFile /config/job-spec.yml

# job-spec.yml — batch ingestion from Parquet on a local filesystem
executionFrameworkSpec:
  name: standalone
  segmentGenerationJobRunnerClassName: org.apache.pinot.plugin.ingestion.batch.standalone.SegmentGenerationJobRunner
  segmentTarPushJobRunnerClassName: org.apache.pinot.plugin.ingestion.batch.standalone.SegmentTarPushJobRunner
inputDirURI: file:///data/raw/events
outputDirURI: file:///data/segments/events
tableSpec:
  tableName: events_OFFLINE
  schemaURI: file:///config/schema.json
  tableConfigURI: file:///config/table-offline.json
pinotFSSpecs:
  - scheme: file
    className: org.apache.pinot.spi.filesystem.LocalPinotFS
recordReaderSpec:
  dataFormat: parquet
  className: org.apache.pinot.plugin.inputformat.parquet.ParquetRecordReader

Inferring a starting schema from sample data — rather than hand-writing every column — is a legitimate shortcut for new datasets; review the result before registering it, because inferred types are guesses.

bin/pinot-admin.sh InferSchema -format JSON -inputDir /data/sample -outputDir /config

---

Transform Functions

Transforms reshape records during ingestion, before they are indexed — the place to flatten nested JSON, cast types and derive timestamps.

"transformConfigs": [
  { "columnName": "event_id",
    "transformFunction": "jsonPath(meta, '$.id')" },
  { "columnName": "event_time",
    "transformFunction": "toEpochMillis(eventTimestamp, 'yyyy-MM-dd''T''HH:mm:ss''Z''')" },
  { "columnName": "raw_meta",
    "transformFunction": "jsonFormat(meta)" }
]

Function	Use
jsonPath(col, '$.path')	Extract a nested field into its own column
jsonFormat(col)	Serialise a map column back to a JSON string
toEpochMillis(col, format)	Convert a formatted string timestamp to epoch milliseconds
groovy / regexpExtract	Custom expressions and string parsing

Doing this at ingestion keeps query-time SQL simple and lets indexes apply to the derived column. Transforms that fail are surfaced in the controller UI's ingestion status — check after every table change.

---

Querying with SQL

Pinot speaks a SQL dialect over HTTP (/query/sql on the broker) or the multi-stage engine, and the results are plain JSON.

curl -s -X POST http://localhost:8099/query/sql \
  -H 'Content-Type: application/json' \
  -d '{"sql":"SELECT count(*) AS changes, count(DISTINCT user) AS users
               FROM events
               WHERE event_time > ago('PT1H')"}' | jq .

-- Time-bucketed trend over the last hour
SELECT DATETRUNC('MINUTE', event_time, 'MILLISECONDS') AS minute,
       COUNT(*) AS changes,
       COUNT(DISTINCT user) AS users,
       COUNT(DISTINCT domain) AS domains
FROM events
WHERE event_time > ago('PT1H')
GROUP BY 1
ORDER BY 1 DESC
LIMIT 60;

-- Top contributors, excluding automated traffic
SELECT user, COUNT(*) AS changes
FROM events
WHERE event_time > ago('PT4H') AND is_bot = false
GROUP BY user
ORDER BY changes DESC
LIMIT 30;

From Python, the Pinot DB-API driver returns pandas frames directly:

import pinotdb
import pandas as pd

conn = pinotdb.connect(host="localhost", port=8099, path="/query/sql", scheme="http")
df = pd.read_sql("SELECT user, COUNT(*) AS changes FROM events "
                 "WHERE event_time > ago('PT15M') GROUP BY user ORDER BY changes DESC LIMIT 20",
                 conn)

Query latency follows from segment design more than from SQL shape: filter on the time column first, avoid SELECT * on wide tables, and prefer pre-aggregated columns to cardinally wild COUNT(DISTINCT ...) over multi-hour windows.

---

Worked Example: Live Event Analytics

A reference architecture for real-time analytics on a public event stream — the same shape applies to application telemetry:

1. Source — an external service publishes a live event stream over server-sent events; a client subscribes and prints or forwards each event.
2. Produce — a small producer consumes that stream and publishes one record per event to a Kafka topic, keyed by the source record ID so per-entity ordering holds. Timestamps are normalised to epoch milliseconds on the way out.
3. Store — Kafka retains the topic, which decouples ingestion from processing and makes replay possible when a transform is fixed.
4. Serve — a Pinot REALTIME table declares the Kafka topic as its stream, extracts nested fields with JSON path transforms, and makes the data queryable within seconds of arrival.
5. Visualise — dashboards query the broker over SQL: per-minute change/volume counts with deltas, distribution across categories, and a drill-down filtered by entity.

The same pipeline supports a batch path: land raw events in object storage, build OFFLINE segments on a schedule, and query the hybrid table so historical backfills and live data appear together. Keeping the raw payload in a column of its own is what makes that backfill possible without re-reading the source.

---

Dashboards over Pinot

Approach	When it fits
Grafana (SQL/Pinot data source or a proxy)	Operational panels beside the rest of the monitoring stack
Streamlit	Internal Python analytics apps; quick to build, needs a query helper for caching
Plotly Dash	Interactive, callback-driven apps with drill-down and live refresh
Superset	SQL-first self-service exploration over the same broker

Two practicalities: dashboards hammering COUNT(DISTINCT) over long windows will not stay sub-second as data grows, so pre-aggregate or shorten the window; and a dashboard should query the broker, never the servers directly, so segment pruning does its work.

---

Operations

• Watch ingestion lag, segment count per table, and query p99 on the broker; these three predict nearly every problem.
• Retention is the storage dial — nothing else reduces a Pinot cluster's footprint as directly.
• Rebalance segments after adding servers so each holds a comparable share.
• Keep schemas and table configs in version control and apply them through the REST API so a rebuild is reproducible.
• Upgrade ZooKeeper, controller, broker and server together; mixed versions across components are not supported.

---

Topics Covered

• Pinot architecture: controller, broker, server, minion, ZooKeeper and segments
• Schema definition — dimensions, metrics, and dateTime columns with correct formats
• Table configuration for REALTIME and OFFLINE tables, and hybrid queries
• Stream ingestion from Kafka and batch ingestion jobs from files
• Transform functions for nested JSON, timestamps and derived columns
• SQL querying over the broker, including time-bucketed aggregation, and the PinotDB connector
• Dashboard integration and day-two operations: retention, rebalancing, lag monitoring