scala.bythebay.io: Hiral Patel, Query Generation Across Multiple Data Stores
Recording: scala.bythebay.io: Hiral Patel, Query Generation Across Multiple Data Stores
you perfect thank you for being here today my name is Hiro Patel senior principal architect and director of engineering at Yahoo for the search in native ads reporting platform okay I've worked on big data various companies in the bay using both open source as well as closed source solutions I'm going to be talking about query generation across multiple data stores today so we're going to cover some of the reasons behind query generation some of the things we did to evolve our query generation process over for our use cases will consider some of the open source alternatives that are out there as well as look at some of our production results and what we're working on next how many of you here know what we mean by OLAP queries awesome most of the room so I'm going to not cover too much in depth here in general for data warehousing the most common data model that's used is the star schema or the snowflake schema here we have dimensions which are the reference information which give context to a fact a fact is a measure of value that represents a fact about a managed system or an entity in the ad system world a dimension is something like the ads that are in the system the campaign's or the ad groups those are the dimensions fact would be something like the number of views associated with an ad or the number of clicks of associated with that ad or the spend associated with that ad so those would be them the facts or measures let's take an example cube definition we have here a performance stats cube which essentially consists of three tables at three different levels we have the ad stats table which is giving us spend across ads the ad group table which gives us spend across the ad ad group and then the campaign stats table so let's say I wanted to run a query to get the spend associated with a particular campaign the best table to run this query on is the table that's going to require me to aggregate the least number of rows to get that data in which case we want to run that query on the the campaign stats table because we're looking at campaign spend so this cube definition has essentially a base table which we call the ad stats table as we define the cube we roll up the ad stats table to the ad group stats table by discarding a single dimension value add a tidy so we can go from the ad stats table to ad group stats table by simply saying okay I'm going to discard one column right and then I can do the same thing with a campaign stats table by by taking the ad group stats table and discard a single column called ad group ID so this is how we build our hierarchy I'm going to be using the terms fields and columns interchangeably I'm going to be using data store and engine interchangeably when I say something is a dimension driven I'm talking about a query where the driving table is a dimension table and a query like that would be gimme all of my ads sorted on a tidy descending so that would be a query where the driving table is a dimension table a fact driven query would be something like give me the top 50 ads on spin descending that would be a factor bin query so why query generation right what's the big deal for our primary use case it was to make it simple for our users to query the data that they needed a simple interface we don't we are used to worry about figuring out what tables that need to court to join with what what are the dimensions what are the fact tables how to construct the query or worry about the kind of join conditions that they need to have either bead either inner join or outer join or left outer join or or or right outer join so we wanted to make it so it was easy for them to query the data that they they wanted by just selecting the fuels that they they wanted and providing a set of filters we also wanted to make it easy to decouple our our interface from the target data store where we ran the query so we can make it make we can easily transition from one data store to another without impacting our users we also have multiple use cases in this case we have you I users who want to manage their entities like ads and campaigns and those are generally dimensioned driven queries they're managing their entities you know changing the aunt Ida or whatever but they also want to see the factual information like the impressions or the spend that they have associated with that ad the other use case we have is users running reports the traditional reporting use case where they say hey can you run this report for me and we say okay sure I will run it asynchronously here's the job ID come back later and check to see if it's done and once it's done they can grab it from from a location that we provide them right so that's an asynchronous use case and those are mostly fact driven queries we also want to just centralized our reporting system we do not want we didn't want to have the business logic being applied across multiple teams and having disparity and how it was being applied right so we wanted a single source of truth for the reporting system so there are many choices when selecting a data store for OLAP queries what do you choose right is it going to be the same data store forever are you ever going to have to change and the answer is going to be yes most likely you're going to have to change at some point our solution generally evolves over time we start simple we used a data store where that where we know best right as the business grows or we start hitting bottlenecks we start looking at other solutions we may find that a certain data store is really good at a certain type of OLAP queries ursus another data store and we want to use both of them right instead of trying to find one data store that solves all of our problems we may find that we have heavy investment in one technology and we don't want to transition off of it like this we want to have the ability to use a separate technology which may be more horizontally scalable and transition incrementally we may also find that the cost of transition is high right so we don't want to migrate from one data store to another and we may want to support both of them concurrently right this is where generating multiple generating queries across multiple data stores makes it easy to evolve our solution over time so when we first started we took a simple approach by by defining an interface for all the things that were interested in for query generation and putting it in an air interface for dimension tape definitions this is what the interface looked like it was pretty trivial and we did something similar for a fact interface you have the fact table name thus columns you're interested in the group by the predicates that you're interested in it's a simple interface you can easily implement it and and add as many tables as you like however we quickly realized the challenges to this approach as the number of cubes that definitions and the number of tables increased it was difficult to scale our solution just wasn't generic enough we didn't have things like derived expression sweet for for example click through rate is a derived expression based on impressions and clicks right we had no way of defining that we had no way of providing hints for optimizations at query construction time like if it's a factor of inquiry I want to use this hint if it's a dimension during query I want to use this hint or or maybe we have a use case where we always want to provide a filter on a column right regardless of what the incoming request wants we're saying okay I want to force this filter on this on this table regardless of what the user wants right so those are constraints or special casing that was not available and so we we knew we needed a different solution so our second approach was to build a sequel like DSL for cube definitions this allowed us to easily add multiple definitions for cubes and of all our feature set here's an example definition where we're defining the performance stats cube with the mapping of the internal columns to the external columns and we're saying that it's based off the ad stats table we also wanted to add the ability to easily define constraints at the column level as well as at the table definition level so we added annotations as you can see we have an annotation here to define foreign key constraint on dimension table for our dimension entities and we also have a table annotation that tells us to use the hive engine for this definition we also wanted to easily support we also wanted to support defining roll-ups easily so this was accomplished by constructing a simple roll up using an existing base table so using the same example we looked at earlier we can construct a ad group stats definition by doing a rip a simple roll up of the ad stats table by discarding the ad ID so it's something that we we talked about earlier and now it's just in dsl for and here we can also define annotations on on grain like if the table is an hourly table or if a table is a daily table this can be useful when we want to determine the time range of the query right if the query is looking for data for the last three hours and and they want to they want hourly data we will use a table that has the hourly grain versus if they want data across the last seven days at the out of the daily grain then we will use the daily grain tables this is another way of optimizing how we're running the query how many rows were going to have to get across to get and get the data we want right we can also define things like high partitioning types oh how the tables are partitioned we can define our partitioning schemes through annotations so what we do when we construct the actual sequel query is inspect the definitions that we've constructed already using our dsl at the request time so when the request comes in we inspect these definitions we figure out what the constraints are we figure out what the special k special special casing that needs to happen for the columns and construct a engine specific sequel based on whichever table we selected to answer this request over time we also found challenges with this approach we found that we were limiting ourselves to selecting a table based solely on the columns that were available so we couldn't make a decision on how to go from select one table will select one data store over another there's no intelligence that says okay this data store is the best data store suited to answer this kind of request there was it was solely based on what columns were available we also found it difficult to extend our existing implementation beyond the two data stores that we support it it was just not generic enough there was the dsl was great but it was also prone to human error we didn't have a lot of error checking and error handling at the definition construction time which led to potential issues that could be found in our function functional testing environment or even in production though it happened rarely also the annotations that we also liked a lot because it allowed us to solve a lot of our special casing use cases but they promoted arbitrary special casing we would find that to annotations may end up actually doing the same exact thing so there's some duplication of code by two different engineers that inserted the the feature at different parts of the pipeline but they did the same thing so this is where we turn to scala we converted everything over to a scholar implementation except for it was all statically typed dsl we retain much of the same functionality we had in our previous version except for we could now rely on scala to help us find the errors in our definitions at compile time as well as construction time so for example here we have a the last column defined here as a derived column called ad status and it is basically a decode of an existing column called status and we reference it using curlies here at construction time we can basically go in there and see does this reference column actually exist in art in our definition right so a construction time we can validate that this definition is correct we can also do things like take a look at the annotations that are available and if the annotation is engine specific I can validate that this annotations applied to a table definition for for the correct engine so if I was for example using a high partitioning scheme for a definition for an Oracle table then it would fail construction it would it would check these things and just tell the developer hey you know you need to fix this it doesn't get past our unit tests we did something similar to 44 facts as you can see here we have foreign key constraints defined for these columns and we can also add construction time check to see are these table references actually do they actually exist in our registry of all the fact all the dimension tables so we can do that at construction time then validate that all of these references are actually correct you we also have the same derived column at the bottom there called a hive derived fact column for video starts I don't know if you can see it very well I guess you can but it is more or less a some of the code which relies on two different columns so one is a dimension column and the other is a fact column so the result of these definitions here for when I call the new fact method here it is essentially defining my base table so this is where I would define the ad stats table it returns a fact builder object and I can use this builder object to then easily add on the the roll-ups that we had talked about earlier just the same way we did with the the sequel base dsl that we looked at earlier so we retain much of the same functionality except for its all statically typed and easy for us to find errors at construction time and and not how has many problems in our funk test environments as well as our production environments so once the definitions are constructed we can convert the builder into a actual cube definition by calling the to public method here what we're providing here is two sets of columns the set of dimension columns and the set of fact columns that are available in this cube definition so in this definition you see the mappings of the internal columns to the external column names and a set of operations that were allowing people our users to apply on these columns we can also specify any forced filters that we want for example if I wanted to always force a filter on source column where the source was always equal to two I can specify that right right here I we can also specify the maximum number of days that we want to allow a query to run on this cube or we can also specify the maximum days look back for example if my maximum days allowed on this key was 30 days and the maximum looked back was 400 days then what I'm basically saying is I can query any 30 days spanning the last 400 days so these kinds of constraints i can define easily at at the cube definition time so i don't have to special case anything i don't have to you know modify my rest api every time i have different constraints for different cube definitions to to get to where we are now what we what we did is we took all the common concepts from the data model and put them into a request model the request model is essentially datastore agnostic it doesn't know anything about what data store is going to be answering this particular request so we took all the common things and put them in one place and have had the query generators a little less complex by allowing them to just work on this request model so the the query generators essentially take this request model and generate a query for the data store they're registered against and similarly the query executor is then also taking the query and generating and executing the query against the data store it is registered against so the executor then more or less populates the result into a realist the idea behind using a registry for the generators and executors is that we can now have multiple implementations for the query generators and executors across each data store and allows which allows us to easily address multiple use cases we can also make our query query generation more intelligent by plugging in cost estimators so i can add in a cost estimator for dimension cost so for a given request i can see hey you know what's the joint cardinality associated with this request and similarly for the fact i can try to estimate the number of rows i'm going to have to a great for this particular request when the request comes in we can take this information and put it into our request model and allow all the components downstream to make decisions on optimizations as they see fit using this information our partition columns are our our definitions are partitioned column aware so if I wanted to plug in a custom custom partitioning scheme I would just need to implement a trade and pass it into my query generator to render our partition columns in order to push changes out to production we added versioning to our cube definitions and as a part of pushing out the new definitions we added the ability to bucket test those those definitions to a very to various sets of users so as we increase the bucket size and feel that the new definitions are good enough we can essentially update the default version to be the new version of the cube definition this allows us to easily make changes in production test them out to a certain set of users and and make sure that there is no user impact let's take a quick look at how a UI request for a you are request a data flow would would look like so we would have an HTTP request coming in with the JSON of describing what the user wants we would validate the object create the request object and use this request object to generate our request model the request model itself jen is generated by inspecting the registered tube definitions and looking at the time providers may be our partitioning scheme requires us to do time zone conversions maybe not all of our users are in the same time zone right so we can get that information from a pluggable time zone provider or a time time provider and we can also get the cost estimate automation estimates for the tables that were interested in so what we do is we create a request model that has all the dimension candidates that can potentially use to answer this request and all the fact candidate tables that can be used to answer this request as well as attach all of the the cost associated with with using those tables for dimensions and facts we can take this information now and generate query pipeline a query pipeline essentially is a way for us to define how we're going to run this request into in the in the case where this is a synchronous query we know that this query is meant to be for the UI so we're going to respond with a JSON response immediately right so the query pipeline essentially is generated to to work with the use case that we're trying to solve for at query time right so the query pipeline is also responsible for determining what's the best data store to to hit which table to hit all that stuff is determined by the query pipeline why would you do us a yes yeah so the question is if I can do a synchronous request why why why you answer it asynchronously right so not all of our requests are synchronous so if you look at the UI use case all of the requests are paginated you're not gonna be able to look at a million ads at a time right so in the case where a user wants to reporter run a report across all of their ads that's going to be an asynchronous report because there's no way we're going to be returning a million rows for for that use case so once once we have a query pipeline we can take this query pipeline which is more or less a program of how we're going to answer this request and schedule it to run on a separate thread which then essentially once executor produces the row list which has all the data that we need and we convert this into a JSON output stream and complete our ASIC response since we use druid and it doesn't support joins out of the box one important use case for us was to be able to query across multiple data stores in order to do this we used our query pipeline abstraction the query pipeline extraction again is how we execute a particular request so this is where we can make the decision on query across multiple engines so this is the use case where we query Oracle for the dimension data and we query druid for the factual data and merge the results in the API itself so let's take an example of getting the top 50 ads with spin descending in that example we would query druid to get the ad ID and the spend information once we have all the entity in for all the entity IDs we can construct an Oracle query to get the ad title and the ad description for those entities once we have that data we can merge it with the data from the first request and serve it out as a JSON output so this is more or less a unique use case for us where we're trying to query across multiple data stores and use each data store for what its best at so why not use one of the open source alternatives like Kyle in or lens I don't as does anyone aware of Kyle in Orleans here all right a few people all right so Kyle in is more or less an an end to end product for managing your cubes it's it generates the data it manages storage it also provides its own anti-sikh oil interface lens is closer to what we've built but it is more like still more like managing a database it also manages your query lifecycle it provides a cube ql interface which is more like a hive ql subset for us our primary use case was to use a library or framework approach so we can easily integrate this into our existing components we don't manage storage and we don't manage the ETL that produces the data our query life cycles are managed by the users of the library themselves we just provide them a the abstractions to execute the query it's there the users of the library who managed actually scheduling the query on a thread and getting the results and doing whatever they want with the results we just provide the extractions and also we wanted to provide a simple JSON interface to our customers that was again our primary use case we we don't want our users to construct any form of sequel and we wanted to easily optimize and customize our generators so optimization for our different use cases was very important for us here's an example JSON input it's pretty simple all we have is the cube that we're interested in the fields that we're interested in and the filters here we're looking at a eight days of data for a particular advertiser so our advertisers or which are mostly our customers worried about a simple JSON input like this and that gets converted into a a query that's joining cross for tables for this for this query because we would have to get the campaign name the ad group name and the ad title so we would join across four tables to get this data here's a set of operations ations that we've built using the cost estimation information that is passed down through the request model for druid we have roughly 4,000 our jury cluster has roughly 4,000 cores that are threaded and we can make decisions on how many cores we want to allocate to running a particular request based on the cost of summation information that is passed through we can also make decisions on what kind of hints we want to send oracle depending on the kind of query it is alright so for high of an Oracle we can also do things like predicate predicates push down so when we're we're joining with large dimensions and some of our dimensions are tens of billions of rows predicates push down becomes a very important optimization so in production we run millions of OLAP queries per day so Optimas up query optimization is not a nice to have it's a must have in order to make sure that we have our essays we are within our SLA s we have 30 plus cube definitions across three data stores hi Oracle and druid our droid cluster itself is segmented on the the idea of tears of service so we have three tiers of service which is supported by our druid cluster and druid is actually one of the few database or data stores which allows you to tear your your storage and tear your your brokers your your actual actual compute our current version of the query generator is three times faster than our previous version which was all written in Java though I'm not saying the Java version was was slow because it was written in Java was just implementation our current implementation is twenty percent less code has more features has better validation and air handling and and this is our this is my plug for Scala essentially yeah query execution is this sorry the question was is the court execution time any different and the query execution time is not different because that's entirely tied to the data stores however because we're making able to make these optimizations on things like you determine the number of course and using the right hints and and and sending the query to the right data store based on how costly of a query it is to run our overall numbers for like ninety-nine percent of time the query is executed at a certain amount of time that those numbers have come down drastically because we're making more intelligent decisions on where to run the query and know the average has gone down yeah no we don't support joints so we we don't so the question is do we support joins across data stores the only place we support joints across data stores is in the UI use case and the UI use case is a special case because all that all the UI results are paginated right so when you're taking a subset of the the results and and sending them through the UI you can afford to do API side joints and that's what we do yes we don't allow direct so you can basically have dimensions that are defined in Oracle and you have facts that are defined in in druid and the framework itself will do the apsi joints for you it did theirs you don't have to do anything but that only works for the UI use case because the async use case is not run in the API side the async use case we have separate workers which use the same libraries to execute the actual async queries right so this is what I mean we've taken a library approach so we can reuse some existing infrastructure of how we process the async use case versus the synchronous response use case we are looking at look we are looking at it yeah yes that's the UI use case because all the UI results are paginated right so we can only you can only show the user a certain number of rows right so in that use case we can we can do cross data store joints on the API side alright so we've also integrated our REST API into a open source data visualization tool called carrabelle that's built by Airbnb which they recently renamed to a superset and it more or less addresses all of our internal reporting needs so some of the some of the work we're working on right now is to add a view support to add fact back chillin support we also want to take into consideration the load on the data store while we're trying to decide on which engine to select so if one data stores you know at a certain percent capacity whereas the other one is underutilized then even though we can answer the query best with this the data store that's loaded I'd rather send it to to the secondary data store because it doesn't have any load right so those are some those are things that we don't currently take into consideration we also want to take into consideration data availability so for example in the use case where we need to do reprocessing and loading into one data store we want to take that that into consideration we want to want to make sure that our users are not hitting a data store where the data is in the process of being deleted or replaced right so we don't currently take that into consideration though we do we do actually I take that back we do take that into consideration for the async use case but not the synchronous use case we're also considering open sourcing this library / framework we're not sure how much interest there would be if this is something that would be of interest to you do let me know so I can take it to to my boss so I've shared our journey for building a general framework for query generation for OLAP queries of course it would not have been possible without all the engineers on my team hopefully I've giving you guys some insights and how how you go about solving the similar problem that in your use cases I can go through an actual cube definition in intellij so you can see what it looks like okay good or yeah so the question is can can I talk more about the fact in view and the fact that joint support work so fact dim view support is basically allowing us to define views inside of the the table definitions itself so an Oracle I can create a view from by doing a union across two tables and define of you doing that instead of having to define the view in Oracle I want to be able to do that directly in my query in my table definitions right so that way I don't have to worry about how that view is actually constructed it the view is more or less just a query across two two tables right so the query generator itself should be able to generate that on its own I just should able to I should just be able to tell the definition itself that hey generate a view across these two tables right so we want to add that support directly into the definition stem cells instead of having to have ddl that goes in there and updates the views in Oracle or or or hive fact fact join support is our use cases where we want to get factual information across multiple facts and join it together right so we don't really support fact fact use cases right now our primary use cases are fact and dim joins or I got one minute so I'm going to skip this questions if no question I can go through the definitions if you'd like okay so this is a definition for our keyword stats cube keyword stats cube is basically performance information about keywords that our customers are interested in as you can see looks similar to what I talked about before a few things that I can tell you about is that we haven't already talked about is this idea of static mappings I can define a static mapping for fields for example the data for price type in the actual table is just a numeric value but externally I don't want to show users a numeric value I can map it back to a human readable value like price type of one is CPC right at the same time this allows me to do a reverse mapping so my customers can tell me okay gimme I want to add a filter where I'm already only interested in where ads where or keywords where the price type is CPC and now we will do a reverse mapping as a part of the query generation process to map CPC and when we're adding the filters and another thing that we didn't talk about here is this idea of aliasing so one thing that you'll encounter is while you're creating tables one table may have the name for a field as a device type ID whereas another table may have the field named as device ID so you may have problems where the naming is not consistent right so how do you get around that problem and so the way that we do that is we just alias it so with this hat we have this idea that although we have a single mapping when we define our cubes for an internal field to an external fuel that internal field name itself may have to be alias depending on the actual table schema definition right so we don't want to go in there and rename tables every time there's a mismatch in the fields right so we can change that or comma do that at the time of query generation okay I think I'm out of time so but what one last thing so here we we can also define constraints around around how when to choose which engine right so we have this dimension cardinality look up which says okay if the dimension cardinality estimate is between 0 and 800 my cost multiplier for a particular engine is one if it's between 801 100,000 then I want to hit drew it if possible essentially so we at definition time we can make these changes and look at how the system performs based on some of these metrics and make these adjustments as as as time goes on all right thank you very much you