scale.bythebay.io: Pavan, Druid Lookups for High Cardinality Dimensions
Recording: scale.bythebay.io: Pavan, Druid Lookups for High Cardinality Dimensions
Good afternoon everyone. Thank you for coming. My name is Pawan. I'm a principal software engineer. I work for the Yahoo Gemini brand at Oath, a Verizon company. Today, I'm going to tell you about a module that we built for Droid to support lookups for high cardinality dimensions. How many people here know Droid? Good, half. I'm going to tell you briefly about Droid and its data segment file, why lookups are needed, why we had to build our own lookup module, its architecture and salient features
Droid is a high performance column oriented distributed data store. It is designed for sub-second ad hoc queries to group, filter and aggregate data. It is ideal for powering user facing analytic applications. We can explore events immediately. They occur. Existing Droid clusters have scaled up to petabytes of data, trillions of events and ingesting millions of events every second. Droid runs on commodity hardware. We can deploy it in the cloud or on premise
We can integrate it existing big data systems like Hadoop, Storm, Kafka, Samsung or Flink. It is a community led project. Droid. To frame our discussion, let's begin with an example data set from online advertising. This data set is composed of three distinct components. Timestamp, where all the queries are centered around time access. Dimensions are string attributes of an event, which are generally used for filtering the data. In our example, we have Advertiser ID, gender and country as dimensions
Similarly, there is metrics column, which is usually numeric. It is used for computation and aggregation. In our example, impression and clicks are the metric columns. The computations are like sum, count and mean. The individual events in our example data set is not very interesting because there can be trillions of such events. However, summarizing this data can yield many useful insights. Druid summarizes these events using a technique called rollup. So, in the example here, there are five raw events
When the rollup is done, it is rolled up into two rows. The compacted version of original raw data looks something like this, which is shown in the bottom of the screen. The rollup aggregation done at the time of ingestion is the minimum granularity that we can explore the data. In this example, I have chosen our as the minimum granularity. That's why we have two rows here. Segments are self-contained containers. It contains indexes in a data structure which is highly optimized for running analytic queries. Here is an example request response
On the left hand side, we have the group by query of the droid where grouping is done on dimensions like for advertiser id, gender, and country. And aggregation is on metrics, impression, and clicks. For the time interval, 2017-0101 to 2017-0102. In the response, we have two rows. For the advertiser id, 12345, country US, and gender male. Similarly, we have one row for gender female. Continuing on our example, what if we want to filter the data based on advertiser status? It's not possible because Druid's immutable segment file doesn't contain status information. This poses a restriction on how we can use Druid with mutable data
Is it possible to overcome this restriction? We have a few options. Number one being client side join where you get the data from Druid and on the client side join the mutable data which is there in another data store. But this has serious overhead. We need to maintain another data store for this mutable data. We need to implement and maintain extra code. And most importantly, this code will have to deal with complicated logic like filtering and limiting the rows. For example, if you want to fetch top 100 advertiser IDs from Druid where status is equal to on, you will not know beforehand how many rows you need to fetch from Druid because you will not know how many of them will have status is equal to on and they will lie in top 100. So, this is going to be very error prone
Another option is using Druid. Another option is using Druid lookups. Lookups is a concept in Druid where dimension values are replaced with new set of values. A key refers to a dimension ID and a value refers to its replacement. In our example, advertiser ID 12345 is the key. Status on is the value. Druid supports two types of lookups. Map lookups
Map lookup, very small lookup that can be passed at query time. It's a static lookup which can be passed during the execution of the query. Another one is globally cached lookups where lookups are loaded from local files or through JDBC. But the default implementation has certain limitations. It is not suitable for high cardinality dimension in the order of millions of rows. And also for the query to run, all the Druid node type like broker, historical and real time will have to load these lookups. Otherwise, it's not possible to solve the query. This poses a serious challenge because certain node types like broker and real time nodes, they run on small memory footprint
And most importantly, the real time nodes create new processes very frequently and they are short lived. Loading entire lookup data is not going to scale. The other limitation is it supports only one key column with a value column. Real updates is not supported, which makes the lookup data to go stale. It doesn't generate any monitoring metrics, which makes it difficult to monitor the application and recover from faults. To overcome these limitations, we worked on creating the new lookup module, which is called Mahadroid Lookups. It is designed for high cardinality dimensions with sub-second updates. It is open sourced under Mahadroid
is a framework for rapid development of reporting APIs. Our colleagues explained about Mahadroid in the previous session. Since it is open sourced under Mahadroid, it has out of the box support for querying Druid using lookups. Let's look at the architecture. The boxes which are orange in color are existing Druid nodes. The Druid queries are sent to broker. Broker then routes the queries to historical nodes as well as real time nodes. Historical nodes will load the historical segment and they make it available for querying
Real time nodes create real time indexes. The data indexed via real time nodes is available for querying. Broker then combines these two results sets and gives it back to the caller. The components which are blue in color belong to our Druid lookup module. When a query with lookup reaches historical node, the module will look into an embeddable persistent key value store called RoxxDB and fetches this value from it. This module using Kafka client can read the messages from Kafka topics as and when they arrive. It keeps on applying all those messages once they are available. The RoxxDB instance itself is created externally using RoxxDB instance creator
It can read the data from HDFS. It ingest the data into the RoxxDB instance using the schema which is provided by us and then uploads that instance onto HDFS. The module on the historical will download this instance and apply all the Kafka messages since the earliest offset and then replaces the current instance with the new one. This replacing the instance is done just to make sure the current one doesn't grow in size with unwanted and unused key value pairs. The module on historical also exposes a rest API which we call lookup service. It is used by broker as well as real time node. As I was telling you broker and real time nodes they are unknown a small memory footprint. They cannot load the entire lookup data
So they use this lookup service exposed by historical and before consuming that service they use a LRU cache to cache the key values. Some of the salient features of our module is it is designed for high cardinality dimensions. We have been using at Oath with more than 120 million unique dimensional rows. It supports multi-column value lookups. To give you an example we were talking about advertiser status. So along with that if you want to look into advertiser time zone, currency or any other dimensions which are mutable like this. Then you need not create new lookup for each of this. You can create only one and it supports the multi-value inside the schema that is provided by us
Real time update is possible using Kafka and it's pretty easy to configure. It supports snapshotting. is very important when we want to restart historical nodes or if you want to deploy new code under the historical. Without snapshotting if you restart a node then what you have to do is you will have to download that instance, RocksDB instance again from HDFS and apply all the messages from Kafka since the earliest offset. This is going to be very time consuming and till such time historical nodes cannot serve any Druid queries. Which affects the availability of the overall cluster. Snapshotting on the other hand stores the instance onto the disk before it gets restarted along with Kafka partitions and the offset which are consumed till that point. Once the node comes back up that instance is already available on the disk
So it can just start using it and the Kafka messages will be read only from those offsets which we have stored. It's highly performant. We have seen less than 25 milliseconds difference between running the query on Android with lookups versus running the entire query on Oracle. It uses RocksDB as an embeddable persistent key value store for fast storage. Using RocksDB is one of the important decisions in our architecture. We wanted an embeddable persistent key value store which can perform under load. So that it complements Druid without losing Druid's power of performance. RocksDB uses a log structured database engine written entirely in C++ for high performance
Keys and values are just arbitrary sized byte streams and it is optimized for Flash or RAM. It scales linearly with number of CPUs so that it works well on the processors with many CPU cores. Here is an example of Druid request response with lookup. This is almost similar to the request response I showed in my earlier slide except for the highlighted part where in the request we are asking for lookup information. And in the response we have the lookup value which is status on. And one good thing about this is it is already integrated with Maha which means it gets out of the box support for querying Druid with lookups. It also provides fallback option to relational database if a Druid query with lookup fails for some reason. It has out of the box query optimizations for filtering on lookup columns
I would like to thank all the contributors and acknowledge their help. Without them this would not have been possible. Thank you.