Druid: A Real-time Analytical Data Store (SIGMOD 2014)

Yang, Tschetter, Léauté, Ray, Merlino and Ganguli — Metamarkets — SIGMOD'14. T1: peer-reviewed, written by the system's architects, with latency and ingestion figures from a named production cluster. The spoke's first source on the real-time serving shape of OLAP, which the founding embedded↔distributed axis does not describe.

The problem it was built for

Metamarkets needed sub-second queries over event streams for 1000+ concurrent users, and found Hadoop was the wrong tool by design: it "does not make any performance guarantees around how quickly that data can be accessed", degrades under concurrent load, and is "not optimized for ingesting data and making that data immediately readable." The paper's verdict on Hadoop — "a great back-office, batch processing, and data warehousing system" — is the sentence that separates this family from the warehouse.

Architecture: node types instead of a cluster

Druid is one system built from four independent node types, deliberately so — "the different node types operate fairly independent of each other and there is minimal interaction among them", so an intra-cluster communication failure barely dents availability. (The name is a joke about this: the Druid class in role-playing games is a shape-shifter.)

Real-time nodes ingest the stream into an in-memory row-oriented index that is "immediately available for querying", periodically persist it to disk in column format, then merge the persisted indexes into an immutable segment and hand it off to deep storage (S3 or HDFS).
Historical nodes load and serve those immutable segments. Shared-nothing, and pointedly dumb: "they only know how to load, drop, and serve immutable segments." They are the main workers.
Broker nodes route queries to the right historical and real-time nodes and merge results, caching per segment.
Coordinator nodes decide which segments live where, from rules in MySQL — load the last month into a hot tier, the last year into a cold tier, drop what is older.

ZooKeeper carries the coordination, MySQL the metadata, and deep storage the durability. A message bus (Kafka) in front of the real-time nodes buffers events and lets several nodes read the same stream for replication; recovery from disk-intact failure takes "a few seconds". One production cluster consumed ~500 MB/s (150,000 events/s, 2 TB/hour).

The storage format, and the thing that makes it a different family

Data sources are partitioned into segments of typically 5–10 million rows, each covering a time interval, identified by source + interval + version. Druid always requires a timestamp column — the paper is explicit that this exists to simplify distribution, retention and "first-level query pruning." That single constraint is what buys the latency: every query is time-bounded before any column is touched.

Columns are dictionary-encoded (strings → integer arrays) and then LZF-compressed, the standard column-store moves this spoke already documents in parquet and orc. What is not standard is the combination with an inverted index over dimension values, which is how arbitrary filter sets stay cheap.

The refusal that defines the shape

Druid had no joins, and the paper defends the absence at length rather than treating it as a gap: "a join query for Druid is not yet implemented… a function of engineering resource allocation and use case decisions more than a decision driven by technical merit." Two stated reasons — scaling joins is "a constant bottleneck of working with distributed databases", and the functionality gain is worth less than the trouble it brings to "highly concurrent, multitenant workloads." Above a billion rows per side, both hash and sorted-merge strategies need pre-join materialization and "complex distributed memory management."

Set against clickhouse and snowflake, both of which do joins, this is the clearest statement in the spoke that the real-time serving family is a different trade, not a weaker version of the same one: give up the general query surface, keep sub-second latency under hundreds of concurrent users.

Numbers, with the conditions attached

Production, Metamarkets' hot tier, early 2014: ~50 data sources, ~10 TB of segments against 10.5 TB of RAM, ~50 billion rows, 1,302 threads on 672 hyperthreaded cores (Xeon E5-2670), memory-mapped rather than heaped. Average query latency ~550 ms; 90% under 1 s, 95% under 2 s, 99% under 10 s. The workload behind those numbers is 60% ordered group-bys, 30% plain aggregates, 10% search and metadata.

Synthetic: TPC-H at 1 GB and 100 GB on EC2, scan rates of 53.5M rows/s/core for count(*) and 36.2M rows/s/core for sum(float). Scaling 8 → 48 cores is linear only for simple aggregations; queries needing broker-level work do not parallelize as well.

Read the comparison with care. The benchmark's only competitor is MySQL/MyISAM, and the paper says why: "we chose not to select another open source column store because we were not confident we could correctly tune it for optimal performance." An honest admission that leaves the headline comparison beating a row store at a column store's job. The production numbers are the useful half, and they describe one company's cluster on hardware where RAM exceeds the working set.

apache-druid · analytical-databases · clickhouse · snowflake-elastic-data-warehouse · storage-compute-disaggregation · parquet · synthesis