Performance optimization for Apache Druid: scaling to 200+ queries per second

Choosing Druid got us the architecture. Making it hold up under real traffic was a separate job. Not long after the real-time store performance product I described in Why Druid went live, concurrent usage climbed past what our first configuration was tuned for, and the number we had to hit was clear: sustain 200+ queries per second without letting latency creep back into “wait for it to load” territory.

There are five levers in a Druid deployment: infrastructure sizing, platform-level cluster configuration, table and schema design, the ingestion spec, and the query and API layer. This post is only about the last three. Infra and platform tuning are a real part of the story, but a different one, and mixing all five together makes it hard to see what actually moved the needle on the query side specifically, and it hides the fact that these changes traded costs against each other rather than being free wins.

Table and storage design

Before any ingestion-spec tuning mattered, we had to decide what the tables looked like. The query layer needed to stop joining, and Druid rewards that by design: flat, denormalized datasources are what its own schema-design guidance recommends, and what its scan-and-aggregate query model is actually built for. So we flattened what had previously lived across a few related tables into fewer, self-contained ones, pulling in the small set of reference values that queries needed instead of joining out to fetch them at read time.

That decision was not free. Flattening a table pulls in columns that used to live in a small, low-cardinality table off to the side, and every one of those columns adds to the cardinality of the rows Druid is now aggregating. That works directly against rollup: rollup earns its savings by collapsing rows that share the same combination of dimension values, and the more distinct combinations a row can have, the fewer rows are left to collapse. We were deliberately trading query-time joins, which cost us on every single query, for higher cardinality, which cost us some rollup efficiency at ingestion time. That trade was worth it for us because joins were the more expensive problem at our concurrency, but it is not automatically the right trade for every workload, and it is worth doing that arithmetic rather than assuming denormalization is a free win.

Ingestion tuning

Most of the query-time cost in Druid is decided long before a query ever runs, at ingestion time, by how segments get built. A handful of changes to the ingestion spec did most of the work here.

• partitionsSpec set to single_dim. Secondary partitioning on a well-chosen dimension keeps rows that get queried together stored together. That locality is what turns a scan into a targeted read instead of a scatter across segments that mostly do not matter to the query. The trade-off is real, though: if the values of the chosen dimension are unevenly distributed, single-dimension partitioning can produce segments well above your target size. It is worth checking that distribution before committing to a dimension, not after.
• rollup enabled wherever it was applicable. Pre-aggregating at ingestion time shrinks the row count the query engine has to touch later. This was the single biggest lever for query throughput, because every query that no longer has to scan raw, unaggregated rows is a query that finishes faster and frees up capacity for the next one. Rollup is not a free discount, either: once rows are aggregated, the row-level detail underneath them is gone for good, and its savings shrink as dimension cardinality rises, which is exactly what the table design above was pushing against. We only rolled up on dimensions where we were confident nobody would need to ask a question at a finer grain than what we kept.
• segmentGranularity and queryGranularity set deliberately, not left at defaults. Getting the time-chunk size right balances segment count against segment size, and queryGranularity was kept at the finest resolution we actually needed and no finer, since anything finer just adds overhead nobody queries for.
• Replication factor set to 3, the same standard most distributed storage systems land on for durability without over-paying in storage and coordination overhead.
• Compaction run after ingestion, so segments produced by streaming ingestion did not sit around suboptimally sized. Compaction cleaned that up on a schedule instead of letting it degrade query performance over time.
• A real data retention policy, so old segments that nobody was querying stopped competing for the same historical node resources as the data actually being served.

None of these are exotic. They are the standard ingestion-tuning checklist. The value was in actually going through it deliberately instead of leaving the defaults in place and tuning the query layer against a foundation that was working against us.

Query and API layer

Nobody queried Druid directly. An API sat in front of it, and that API layer turned out to have as much headroom as the ingestion spec did.

Response format: JSON to CSV. Druid’s default query response repeats every column name on every single row, which is fine for a handful of rows and wasteful for the wide, high-row-count results our aggregations were returning. Switching the API’s Druid queries to request a CSV response instead sends the column headers once and just the values after that. For our result shapes, that roughly halved the bytes moving over the network for a typical query, which mattered twice: less time spent transferring data, and less time spent serializing and deserializing it on both ends. CSV only worked because what we were sending back was flat, tabular, column-typed aggregate data, which is exactly what Druid’s query results look like. It has no notion of nested structure and no per-value typing, so it would not have been an option if any result needed to carry nested or variably-typed fields.

Joins pushed out of Druid. Druid’s query engine is built to scan and aggregate at scale, not to join, and the table redesign above did most of the work of removing joins entirely. For the small set of cases that still needed one, we moved that step out of Druid and resolved it at the API layer instead, after the Druid query came back. That is a relocation of the cost, not a deletion of it, and it was only a net win because what moved to the API layer was small and already cheap to resolve there. If that side of the lookup had been expensive too, we would have just moved the bottleneck instead of removing it. Keeping Druid doing only what it is good at, and pushing everything else to a layer actually built for it, was worth more than any individual query tweak, but only because we checked that the new location was cheaper than the old one.

What it added up to

Between the three, we cleared the 200 queries-per-second target with typical response times still sitting in the low hundreds of milliseconds, even at peak concurrency. The response format change alone was the fastest win to ship: it required no re-ingestion and no schema change, just a different flag on the outbound query. The table design and ingestion changes took longer to pay off but mattered more at sustained scale, since they lowered the cost of every query that followed rather than just the size of its response.

Lessons

None of these changes were free, and that is the actual lesson, not the checklist itself. Rollup traded away row-level detail for fewer rows to scan. Denormalizing traded query-time joins for higher cardinality, which quietly ate back some of what rollup bought us. Pushing the remaining joins to the API layer traded Druid-side cost for API-side cost, and only paid off because that side was genuinely cheap. Table design, ingestion, and the query layer were not three independent wins; they were one set of trade-offs made across three layers, and the work was figuring out which costs we could afford to pay and where, not finding costs we could avoid paying altogether.