How Netflix uses Druid for Real-time Insights to Ensure a High-Quality Experience

Why Netflix built a real-time metrics pipeline on Druid

Netflix pushes software updates to hundreds of device types continuously. Each change carries risk: a regression might only surface on a specific Smart TV model, an older Android build, or users in one region. To catch those problems before they spread, the company needs to measure playback and browsing behavior from real device logs the moment events happen.

The scale of that monitoring is substantial. Playback devices generate more than 2 million events per second, which works out to over 115 billion rows per day. Those rows carry high-cardinality dimension data—device type, app version, country—that must remain queryable within tens of milliseconds so teams can isolate and respond to anomalies as they occur.

How Druid organizes high-cardinality data

Netflix chose Apache Druid because it fits this specific workload: high ingestion rates, high dimensionality, and fast query requirements. Druid is not a relational database; the closest analog to a table is a datasource. There are no joins, so every column you might want to filter or group by must be present in the datasource itself.

Each datasource contains three column types. A timestamp column is the primary partition key—everything in Druid is organized by time. Dimensions are fields used for filtering, grouping, and querying. Metrics are numeric values subject to aggregation. The trade-offs of removing joins and keying data by time allow Druid to scale to trillions of rows while returning query results in tens of milliseconds.

Segment-based storage and query fan-out

Druid partitions stored data into configurable time chunks; Netflix uses one-hour chunks for its metrics. Each chunk's rows live in one or more segments, with configurable upper bounds on row counts and file sizes. When a query arrives, it fans out to every node holding segments within the requested time range. Each node processes in parallel and returns intermediate results to a broker node, which performs the final merge and aggregation.

Rollup at ingest and after the fact

Druid supports rollup during ingestion as a form of pre-aggregation. Any rows with identical dimensions that fall within the same minute—Netflix's chosen query granularity—are combined by summing their metric values and incrementing a counter that tracks how many source events contributed to the row. This can reduce row counts by orders of magnitude, though it eliminates the ability to query individual events and limits results to the configured granularity.

Rollup efficiency drops as dimension cardinality rises. With many parallel indexers, identical rows rarely land in the same ingestion task, so rollup happens imperfectly. To recover those gains, Netflix schedules a compaction task for each time chunk after all its segments have been handed off to Historical Nodes. The compaction job fetches segments from deep storage, runs a map/reduce-style pass, recreates the segments, and achieves a near-perfect rollup—roughly a 2x improvement in row count over the initial ingest.

Handling the real-time edge cases

Compaction only succeeds if all data for a time chunk has actually arrived. Three safeguards keep that true at Netflix. First, any data arriving too late is discarded—the real-time system has no use for it. Second, compaction is scheduled with a deliberate delay so segments have time to complete normal handoff. Third, when the compaction task starts, it checks segment metadata to see if any segments are still being written or handed off; if so, it waits a few minutes and retries.

These checks matter. Without them, Netflix found that compaction could overwrite segments that were still receiving writes from indexers. The newer compacted segments would take precedence because they carried a higher version, effectively deleting data that had not yet finished handing off.

From log streams to queryable metrics

Ingestion does not insert individual records into a datasource. Netflix reads metrics from Kafka, using one topic per datasource, and relies on Druid's Kafka Indexing Tasks. These tasks spawn multiple indexing workers distributed across Middle Managers. Each indexer subscribes to its topic, extracts fields according to an Ingestion Spec, and accumulates rows in memory.

Newly created rows are queryable immediately. For time chunks where segments are still filling, queries are served directly by the indexer. Because indexing tasks carry both ingestion and query duties, Netflix offloads segments to Historical Nodes promptly once thresholds are met—either enough accumulated rows or a maximum open-segment duration. The indexer writes the segment file to deep storage, alerts the coordinator, and the coordinator assigns it to Historical Nodes. Once loaded, the segment is unloaded from the indexer and future queries target the Historical Nodes instead.

Querying through an Atlas translation layer

Druid supports both Druid SQL and native queries, which are JSON submissions to a REST endpoint. Netflix primarily uses native queries, not because there are no alternatives, but because its dashboards and alerting systems were designed around Atlas, Netflix's internally developed and open-sourced time-series database.

To avoid rebuilding every tool, Netflix added a translation layer that accepts Atlas Stack queries, rewrites them as Druid queries, issues the request, and reformats the result set as Atlas output. Existing dashboards and alarms now work against the Druid datastore without modification or additional training for the teams using them.

Benchmarking Cluster Tuning

To evaluate configuration changes on cluster nodes, we ran repeatable, high-rate benchmark queries designed to isolate specific parts of the system. These targeted tests let us measure response times and throughput while checking for improvements or regressions. For example, queries against the most recent data hit only Middle Managers; longer-duration queries over older data targeted Historical nodes to validate caching setups; and queries with high-cardinality group-bys tested result-merging performance. We iterated on these benchmarks until query performance met our targets.

The most effective tuning levers were buffer sizes, thread counts, query queue lengths, and memory allocated to query caches. But the single biggest win came from introducing a compaction job that re-compacts poorly-rolled-up segments with perfect roll-up. Additionally, enabling caches on Historical nodes yielded major benefits, while caches on broker nodes were nearly useless for us — almost every query misses the broker cache because it includes real-time data that is always arriving.

Ongoing Operations

After several tuning iterations, Druid has lived up to our initial expectations as a capable, usable system for our data rates. The work is not done, though: ingestion volume and rates keep climbing, as do query count and complexity. As more teams realize the value of this detailed data, constant additions of metrics and dimensions push the system harder, so we must continuously monitor and adjust to keep query performance in check.

Today we ingest over 2 million events per second and query over 1.5 trillion rows to extract detailed insights into how users experience our service. This capability helps sustain a high-quality Netflix experience while supporting ongoing innovation.