This page is meant to assist users in designing a schema for data to be ingested in Apache Druid (incubating). Druid offers a unique data modeling system that bears similarity to both relational and timeseries models. The key factors are:
The rest of this page discusses tips for users coming from other kinds of systems, as well as general tips and common practices.
(Like Hive or PostgreSQL.)
Druid datasources are generally equivalent to tables in a relational database. Druid lookups can act similarly to data-warehouse-style dimension tables, but as you'll see below, denormalization is often recommended if you can get away with it.
Common practice for relational data modeling involves normalization: the idea of splitting up data into multiple tables such that data redundancy is reduced or eliminated. For example, in a “sales” table, best-practices relational modeling calls for a “product id” column that is a foreign key into a separate “products” table, which in turn has “product id”, “product name”, and “product category” columns. This prevents the product name and category from needing to be repeated on different rows in the “sales” table that refer to the same product.
In Druid, on the other hand, it is common to use totally flat datasources that do not require joins at query time. In the example of the “sales” table, in Druid it would be typical to store “product_id”, “product_name”, and “product_category” as dimensions directly in a Druid “sales” datasource, without using a separate “products” table. Totally flat schemas substantially increase performance, since the need for joins is eliminated at query time. As an an added speed boost, this also allows Druid's query layer to operate directly on compressed dictionary-encoded data. Perhaps counter-intuitively, this does not substantially increase storage footprint relative to normalized schemas, since Druid uses dictionary encoding to effectively store just a single integer per row for string columns.
If necessary, Druid datasources can be partially normalized through the use of lookups, which are the rough equivalent of dimension tables in a relational database. At query time, you would use Druid's SQL LOOKUP function, or native lookup extraction functions, instead of using the JOIN keyword like you would in a relational database. Since lookup tables impose an increase in memory footprint and incur more computational overhead at query time, it is only recommended to do this if you need the ability to update a lookup table and have the changes reflected immediately for already-ingested rows in your main table.
Tips for modeling relational data in Druid:
(Like OpenTSDB or InfluxDB.)
Similar to time series databases, Druid's data model requires a timestamp. Druid is not a timeseries database, but it is a natural choice for storing timeseries data. Its flexible data model allows it to store both timeseries and non-timeseries data, even in the same datasource.
To achieve best-case compression and query performance in Druid for timeseries data, it is important to partition and sort by metric name, like timeseries databases often do. See Partitioning and sorting for more details.
Tips for modeling timeseries data in Druid:
(Like Elasticsearch or Splunk.)
Similar to log aggregation systems, Druid offers inverted indexes for fast searching and filtering. Druid's search capabilities are generally less developed than these systems, and its analytical capabilities are generally more developed. The main data modeling differences between Druid and these systems are that when ingesting data into Druid, you must be more explicit. Druid columns have types specific upfront and Druid does not, at this time, natively support nested data.
Tips for modeling log data in Druid:
Druid can roll up data as it is ingested to minimize the amount of raw data that needs to be stored. Rollup is a form of summarization or pre-aggregation. Columns stored in a Druid datasource are split into dimensions and measures. When rollup is enabled, any number of rows that have identical dimensions to each other (including an identical timestamp after queryGranularity-based truncation has been applied) can be collapsed into a single row in Druid.
In practice, rolling up data can dramatically reduce the size of data that needs to be stored, reducing row counts by potentially orders of magnitude. This storage reduction does come at a cost: as we roll up data, we lose the ability to query individual events.
You can measure the rollup ratio of a datasource by comparing the number of rows in Druid with the number of ingested events. One way to do this is with a Druid SQL query like:
-- "* 1.0" so we get decimal rather than integer division
SELECT SUM("event_count") / COUNT(*) * 1.0 FROM datasource
In this case, event_count was a “count” type metric specified at ingestion time. See Counting the number of ingested events below for more details about how counting works when rollup is enabled.
Tips for maximizing rollup:
queryGranularity at ingestion time (for example, using PT5M instead of PT1M) increases the likelihood of two rows in Druid having matching timestamps, and can improve your rollup ratios.For more details about how rollup works and how to configure it, see the ingestion overview.
Druid always partitions your data by time, but the segments within a particular time chunk may be partitioned further using options that vary based on the ingestion method you have chosen.
In general, partitioning using a particular dimension will improve locality, meaning that rows with the same value for that dimension are stored together and can be accessed quickly. This gives you better performance when querying that dimension, including both filtering and grouping on it. Partitioning on a dimension that “naturally” partitions your data (such as a customer ID) will also tend to improve compression and give you a smaller storage footprint. These effects will be maximized by putting the partition dimension first in the “dimensions” list of your “dimensionsSpec”, which also tells Druid to sort data segments by that column.
Note that Druid always sorts rows within a segment by timestamp first, even before the first dimension listed in your dimensionsSpec. This can affect storage footprint and data locality. If you want to truly sort by a dimension, you can work around this by setting queryGranularity equal to segmentGranularity in your ingestion spec, and then if you need finer-granularity timestamps, ingesting your timestamp as a separate long-typed dimension. See Secondary timestamps below for more information. This limitation may be removed in future versions of Druid.
For details about how partitioning works and how to configure it, see the ingestion overview.
When dealing with high cardinality columns like user IDs or other unique IDs, consider using sketches for approximate analysis rather than operating on the actual values. When you ingest data using a sketch, Druid does not store the original raw data, but instead stores a “sketch” of it that it can feed into a later computation at query time. Popular use cases for sketches include count-distinct and quantile computation. Each sketch is designed for just one particular kind of computation.
In general using sketches serves two main purposes: improving rollup, and reducing memory footprint at query time.
Sketches improve rollup ratios because they allow you to collapse multiple distinct values into the same sketch. For example, if you have two rows that are identical except for a user ID (perhaps two users did the same action at the same time), storing them in a count-distinct sketch instead of as-is means you can store the data in one row instead of two. You won‘t be able to retrieve the user IDs or compute exact distinct counts, but you’ll still be able to compute approximate distinct counts, and you'll reduce your storage footprint.
Sketches reduce memory footprint at query time because they limit the amount of data that needs to be shuffled between servers. For example, in a quantile computation, instead of needing to send all data points to a central location so they can be sorted and the quantile can be computed, Druid instead only needs to send a sketch of the points. This can reduce data transfer needs to mere kilobytes.
For details about the sketches available in Druid, see the approximate aggregators page.
If you prefer videos, take a look at Not exactly!, a conference talk about sketches in Druid.
If the user wishes to ingest a column as a numeric-typed dimension (Long, Double or Float), it is necessary to specify the type of the column in the dimensions section of the dimensionsSpec. If the type is omitted, Druid will ingest a column as the default String type.
There are performance tradeoffs between string and numeric columns. Numeric columns are generally faster to group on than string columns. But unlike string columns, numeric columns don't have indexes, so they can be slower to filter on. You may want to experiment to find the optimal choice for your use case.
For details about how to configure numeric dimensions, see the Dimension Schema page.
Druid schemas must always include a primary timestamp. The primary timestamp is used for partitioning and sorting your data, so it should be the timestamp that you will most often filter on. Druid is able to rapidly identify and retrieve data corresponding to time ranges of the primary timestamp column.
If your data has more than one timestamp, you can ingest the others as secondary timestamps. The best way to do this is to ingest them as long-typed dimensions in milliseconds format. If necessary, you can get them into this format using transform specs and expressions like timestamp_parse, which returns millisecond timestamps.
At query time, you can query secondary timestamps with SQL time functions like MILLIS_TO_TIMESTAMP, TIME_FLOOR, and others. If you're using native Druid queries, you can use expressions.
At the time of this writing, Druid does not support nested dimensions. Nested dimensions need to be flattened. For example, if you have data of the following form:
{"foo":{"bar": 3}}
then before indexing it, you should transform it to:
{"foo_bar": 3}
Druid is capable of flattening JSON, Avro, or Parquet input data. Please read about flattenSpecs for more details.
When rollup is enabled, count aggregators at query time do not actually tell you the number of rows that have been ingested. They tell you the number of rows in the Druid datasource, which may be smaller than the number of rows ingested.
In this case, a count aggregator at ingestion time can be used to count the number of events. However, it is important to note that when you query for this metric, you should use a longSum aggregator. A count aggregator at query time will return the number of Druid rows for the time interval, which can be used to determine what the roll-up ratio was.
To clarify with an example, if your ingestion spec contains:
...
"metricsSpec" : [
{
"type" : "count",
"name" : "count"
},
...
You should query for the number of ingested rows with:
...
"aggregations": [
{ "type": "longSum", "name": "numIngestedEvents", "fieldName": "count" },
...
If the dimensions field is left empty in your ingestion spec, Druid will treat every column that is not the timestamp column, a dimension that has been excluded, or a metric column as a dimension.
Note that when using schema-less ingestion, all dimensions will be ingested as String-typed dimensions.
One workflow with unique IDs is to be able to filter on a particular ID, while still being able to do fast unique counts on the ID column. If you are not using schema-less dimensions, this use case is supported by setting the name of the metric to something different than the dimension. If you are using schema-less dimensions, the best practice here is to include the same column twice, once as a dimension, and as a hyperUnique metric. This may involve some work at ETL time.
As an example, for schema-less dimensions, repeat the same column:
{"device_id_dim":123, "device_id_met":123}
and in your metricsSpec, include:
{ "type" : "hyperUnique", "name" : "devices", "fieldName" : "device_id_met" }
device_id_dim should automatically get picked up as a dimension.