Adding Custom Optimizations to Catalyst...

Scale By The Bay 2018: Russell Spitzer, Adding Custom Optimizations to Catalyst....

Recording: Scale By The Bay 2018: Russell Spitzer, Adding Custom Optimizations to Catalyst....

yeah so hi everybody thanks for coming out me on a Saturday as said before I'm Russell Spitzer I work at data stacks and main thing I work on is Cassandra and SPARC and the integration between the two of them so there is the data stack SPARC Cassandra connector which is one of our biggest open-source components which basically lets you take your data from Cassandra you work with it in SPARC and of course take any data that you have access to in SPARC and put it back into Cassandra so in case anyone hasn't used SPARC before just a quick description SPARC is you know a tool for working with big data actually just stole right out of the Apache Hadoop homepage it's description of Apache Hadoop because it's basically the exact same description it's a framework that lets you write applications that do distributed work without having to worry about the nitty-gritty of what happens when something breaks what happens when a node goes down how am I actually splitting up my data and all this kind of stuff so it kind of takes care of a lot of that for you so basically working on your Big Data using high-level extract abstractions you know things that are very familiar to us when we're working with Scala so you have a very similar collection API except instead of a map only working on a single machine when you map and spark it's going to work on you know as many machines those are in your spark cluster so one of the newer additions all this is not too new anymore is that sparks biggest direction now is is taking SQL and SQL like statements and turning that into distribute at work so for example you might have a select count from some source which might be in Cassandra it might be some other database might even be just an old Oracle store somewhere else and what it will do is take these rather simple high-level descriptions of what you want to do and turn it into distributed work on whatever your system happens to be you know whether or not you're on the cloud or your in-home data center or on your own dev box you know whatever you're doing it's going to take your little description and change it into a larger distributed computation so how this happens is a little bit of a dark art so for example you take this query I'm taking a select from one table joining it with another table gets fed into this thing called catalyst so catalyst is the module within spark that does all of this parsing and optimization and at the end you end up doing distributed work so what's really happening in here you know a lot a lot of people take a look at this and you just kind of pass like I don't really need to know what's happening on the inside what's really more important is that I get this output that I want but today we're going to talk a lot about how the internals of this work so we can figure out how we can actually change them and have it behave in different ways that might be more beneficial to the specific data sources so let's take a little example here this is some actual compiling Scala spark code and imagine we have these two tables and we want to provide perform a join between the two of them so we basically make a reference to each of those tables we can do some filters we're looking for you know in this particular join we're looking for things where the visit ID is greater than one where the cost was more than 10 we want to join these two two databases because one has which dogs a person owns and the other has some information about their billing when they went to the vet or not we need to match this together so we can figure out which dogs were at which of has visits or something like that so we do this description here which looks a lot like SQL there's actually an SQL description of this that you can write as well but this is the scala api and in the end it's going to give us the results we want but under the hood what's happening so the first thing that that happens is that code that scala we've written gets transformed into a tree a parsed tree so all of those operations end up being represented as specific classes so we've got you know this join thing these filters these relations that we're pulling data from and we can actually look at all these and kind of classify them by type so this tree says in order to do this amount of work the what we need to actually do is we have to perform a join the join requires this owner column number nine and owner column 20 to 24 to get those columns we need to do projections off of some table that we're reading information from and we have these filters so these filters you know happen in certain places and this is basically a big explanation of what needs to happen to get the answer to this request so it's pretty important to realize that this at this level at this stage of optimization really what we're focused on is what needs to happen not how does it need to happen because actually figuring out the how of it is going to be a little more complicated and once we're at that stage it's a lot harder to make optimizations so each of these nodes in this plan is what we call a logical node because it is a logical description rather than a physical description of what of what is happening these nodes are built pretty simply so they have you know a a class which extends logical plan in this case this is filter filter extends logical plan each logical plan node has an output tells us which columns are coming out of it which columns are produced by this and we have a set of children which is what are the things that happen before this so for example this filter is applied to another tree and that tree content consists of a filter and a relation so inside of that we still have to describe this this actual filter itself which is that we're looking at this cost column and we're looking for it to be greater than 10 so that actually is what we call an expression which is also going to become a tree and this is in another class and the expressions so all of these extend expressions now and they can have things like this reference right here is a greater than and the greater then of course has two children one for each side in this case one of these is that a reference to a value in a tree or a value in a table and the other is just the literal number 10 so we have this tree and inside of our tree we have more trees and but basically everything is structured where we have this very clear definition of of what needs to be done to accomplish this logical part of the query so looking at it from a much higher level view every logical node is a definition of some inputs that get changed by some set of expressions that produce some set of outputs and if you know these input and outputs you can make a lot of you can make a lot of choices about the order of things you can decide whether or not a certain node can be in a certain place or whether you can move it up or down within that tree because for example if I'm doing a filter well there's no reason that one filter has to happen before another filter we can always swap their order and get the same result so if we can do those sorts of things we need to have rules that explain how do we swap these things how do we move them around how do we check whether any of this is valid so this logical plan then goes through a set of optimizers so there's two phases which change the the logical plan tree but they change it into yet another logical tree so in this case these phases are the analyzer phase in the optimizer phase and we'll just go briefly through what those do so we can get an idea of why we might want to add things in those places so to change a logical tree we use something called a rule and this is just the definition of the rule class and spark and what you'll see is what the rule does all all we have to do is we take in one tree and we output one tree that's all a rule has to do it can do whatever one wants inside of that but what it really needs to do is take in a tree do something to it and output another tree so rules can be applied in a bunch of different ways so for example this is a bit of code showing you that there's a certain set of rules that they've labeled substitution the c2e substitution windows substitution eliminate unions all of these things and the way catalysts can apply these rules is based on a parameter here here it's fixed point fixed point means keep applying this rule over and over and over until the output matches the input there's other options you can set it to only apply a rule one time or only apply a rule n times but in this case we basically say we're just keep applying the rule over and over until it doesn't work anymore so this is good for things like recursive rules where you know we might eliminate for example we might eliminate a union that then provides us with a new pair of expressions that we can also eliminate so we would keep applying it over and over till we've gotten out of that that state so just to go over examples of what analyzer rules look like this is that first stage of logical plan optimization here's a list of all the ones that are currently in spark 2.3 don't worry about that the thing in general is that analyzer rules are all about saying is this even possible are the things you're calling real so for example you called a UDF in the middle of your code that you know is called bob's UDF 5 right so it needs to check within the catalog it needs to check within spark to say is that a real function that exists in this scope can I actually do something with that or you might recall on a table and to call on a table it has to go then check in the catalog and see does this table actually exist so it's all those kinds of rules that's like are you looking for columns that exist are you looking for tables that exist so this phase is basically our first sanity check to make sure that this query is at least reasonable at least we're talking about things that the engine knows how to handle here's an example of 4 I've done a nice clickbait title here just so you might want to look at 4 it's not really anything special but you know like I was saying we're making sure that you know we have relations that exist inside of our catalog we're making sure our functions are real we're making sure that you know if we use the star or we've alias the column as something else that that's valid you know so that if we refer we select a column and maybe rename it as something else we select call an X and rename it to Y we need to make sure later on that all of our Y references can be mapped back to a real X so that's the kind of checks that are happening in this analysis phase so here's an example of one of those rules in action so this is that eliminate unions I was talking about before so this is basically just saying if I'm doing something if I'm trying to do a union and I'm trying to do a union and there's only one thing in my union then I really don't have to do anything at all so take that tree that has a child and just take the child so you can imagine now this is why I would want to apply this rule over and over and over again because if for example I say I would like to take the union of this and then I'd like to take the union of that and you can keep wrapping this over and over and over but logically it's all the same thing a union of one object is the same as just that object so you can keep unwrapping them until you end up with a single child so this is the actual code for that eliminate union like I said basically says if you've got one child then take the child that's it so the more exciting rules are optimizer rules so this is that second stage of the logical plan analysis so the logical plan now says we want to make things better we aren't just trying to check whether or not functions exist or catalogs or tables exist now we're really interested in can we do it better without actually knowing how we're going to do it so I know that might sound a little confusing but let's just go over some examples so again I have another clickbait title here's five examples for example if we want to limit if we have a limit inside of our logical plan there's almost no cases and where if we can move that limit closer to the actual read or closer to the source of data it's almost always going to be better so we have a rule that says if you have a note that's limiting the amount of results keep moving that down as far as you can in the plan so it happens as quickly as possible another one like constant folding if you're in your application you do a request that's select whether or not column Fah our column X is greater than one plus one well you know computers are very smart and good at math so they might as well just do greater than two because Y calculate one plus one every single time you're comparing a row so this would take two different nodes that are part of that addition so it probably would be add literal one add literal one and combines them into literal two another one combine filters imagine you have multiple filters that are on a single column it probably helps to combine them all into a single logical node so that things down the the road can actually take that full set of expressions and do something with them rather than assuming there's just one at a time cost based joinery reorder this is one where it can actually look at the underlying stats if you've looked into the cost-based optimizer and spark this is the node that basically says okay this joint should probably happened before that join because this join is going to require less data and will mean that the second join will end up being a lot more efficient and at the end you know there's a rule like simplify cast so casts are UDF if I have to change from one data type to another data type that can be expensive but if I already have an integer and I try to cast that into an integer well just remove that just remove that entire operation so we can actually save time by getting rid of things that don't need to be done so here's an example of how that unneeded cast rule actually works so it's again like the other rules we looked at it takes a logical plan outputs a logical plan and this one just looks and whether or not you're casting a column to a data type and if the columns data type is the data type you're casting it to we remove that whole node there's some other specialty things here because you have in complex types this becomes a little more complicated but this is I just wanted to show this part because it's very easy to understand that if we have the same data type that we are we're looking at just get rid of it so for example this projection with a cast inside of it can be changed just to the projection of that column because in the case that a is an int type so we just remove that extra bit of work so again these are all talking about what needs to happen what kind of operations need to happen before our query actually can be executed the hard part of this of course is none of this talks about how to do this distributed lis none of this says actually how to do the the math how to do the programming that actually will take our rows put it through all of this and give us a result set now for that we have to also first we apply all of our rules and we got our our optimize plan like I said we combine our filters all that kind of stuff but we need to turn this into a description of what's happening what do we actually have to do does this relation mean that we actually have to read data out of Cassandra or does this mean we have to get data out of JDBC does that mean we start a connection pool and start pulling out rows one at a time like we actually have to figure that out before we can do any work so that's what that the last transformation inside of catalyst is and that's the application of strategies this is the planning stage so the planning stage is different than the other stages before this because it takes us from a logical plan into a spark plan spark plan is part of a physical plan which is a description of the actual code we will be running one row at a time to actually get results so an example of what this actually means in practice is something like this so for example if I have a join between a column a and a column B well we know what a join looks like when it's done but how do we get to that phase so in spark there's a bunch of options so for example it could do a sort merge join this is where it takes the tables and it writes out many sub files and those sub files get shuffled all around and then they get merged back together and we do a join that way with this big distributed shuffle operation it's very expensive we have another option called broadcast join a broadcast join in spark takes one side of the join takes it into a map parallel Liza's that to every executor and then just does a lookup within that map so deciding which of these you're going to actually do you can see this is the actual physical thing what are we actually going to do to get the answer is pretty important choice we have to decide which of these we do and then on the other hand we have these relations so remember I said we had that logical relation at the bottom that refers to something in the catalog or some other kind of source well we have to turn that into code that actually takes data out of that source so for example logical relations if you're familiar with spark you can do spark read format CSV or read format park' or read format Cassandra that will then get transformed into a physical piece of a physical reference to one of the data sources that actually reads data out of out of those various external third-party things so one little issue here you might have noticed is that if we're writing these plans that take a logical plan and turn them into a spark plan we kind of have our problem applying them piecewise because if we want to apply something over and over again we can't really do that because once we've applied the rule once we should end up with a spark plan we shouldn't end up with a logical plan so we can't apply the same rule over and over again well the way that spark gets around this is that one of the things you do when you're writing a rule so this is a lot of the internals is anything that you're not taking care of in your planner rule you wrap around with a plan later so plan later makes this into a physical node but it's a physical node that actually just gets wiped cleaned every single time we go through it so we can actually do something over and over again so say for example I don't know how to deal with the join in the rule that I've written but I do know how to deal with something else so I can handle that part and change that into physical nodes and then for all the things that I don't understand that I don't know how to work with I just wrap them in a plan later so then that will get back into the planner again and the planner will be able to unwrap that and try again with other rules that are within that set of planning rules so here's a quick example of what's inside these spark plan notes so again we actually have a node that also just says filter at the front but this is execution filter exe C so inside of the spark code base if you take a look all of these physical nodes have these Exe C preferences or suffixes and they extend a whole lot of other stuff and this is where a lot of the cool things in spark actually come into play so we'll go through a little bit of that so for example since this node has is a filter Exe it has a do execute method so this is where we actually have the code to deal with a row you'll notice we can take an RDD of rows here we can actually do something on them the first thing it does is of course call its children and has them do whatever they need to do and then it takes the iterator that comes out of that and you'll notice it takes the predicate that is inside of this makes an operation on it and then iterates all the rows coming through with the predicate so basically this is the literal code that will be evaluated every single time you go through a filter step inside of your inside of your job but you've probably heard a little bit about how SPARC now has a whole lot of cool code gen stuff it's actually gonna compile new code when you write a plan so how does that come into play so if you notice up here there's another thing extended here called code gems support so code gen support gives you the opportunity within this execution node to give these two other methods do produce and do consume which basically let you say what code needs to be generated if this node is in place in code gen is enabled so instead of saying how did it work with a row it actually expects you to produce a string which literally has the code that would be run you know that will be added to the compilation for the large code gen section so again this is a lot of the internals but this is how it actually happens so if you put a whole lot of these notes together it will use all of these do produce and do consume notes to get strings of code that it will put into a class file and then compile so that's how we actually get the how to work with it even when we're doing code gen again this how has to be fully explained here we have no other stages after this which will say how to work with a row how to actually deal with the data we're working with so this is the the bottom line whatever is in here is what actually happens so what kind of planner rules are there so there's a whole lot of different kind of planner rules this is actually it this is the whole list I'll just go over a few of the ones that are really relevant to me as someone who writes the SPARC Cassandra connector filter pushdowns is one of the planning rules because there's a point now where we have this logical relation that says we need to do a filter on this column but we also have this relation that says we're pulling data from somewhere else this is the case where we can actually look and say oh okay that particular predicate can be handled inside of that source now that we know that the source is actually Cassandra cause Cassandra has certain rules about what filters you can use other things like column pruning if we can tell that we note need all of the columns out of the source we can actually tell that base relationship don't read all those data don't read all those different columns and of course all of the other stuff where we write how things actually get written to the the data source and how we make the connection pools and all of that all of that stuff gets happened when we apply these rules so we go from our previous optimized logical plan into this physical plan so this is the physical representation of the same plan we were looking at before you'll see a lot of this is a little different there are a few things that you probably want to see when you're doing your own requests and you do and explain you'll notice that you've got these stars on some of these methods the Stars all represent a place where code gem is happening so this shows us that it's not using those do execute methods instead it's actually producing that code and compiling a new class that will run through my data it also shows us now a lot of important information about what filters have been pushed down to the database what are the actual rows were reading out of the database all that kind of stuff so we actually get a much clearer prediction of what's actually happening and in this case we can tell that the the join we're doing here is a sort merge-join so it's decided that both of these tables are large enough based on the size estimates that they've given that it does not want to do a join based on a broadcast so you can actually tell why this would be much slower than a different plan based on the operations that it's doing so what I wanted to talk about given all of this information is how can we make this better how can we add into this framework so one thing I know about Cassandra is Cassandra has really fast key lookups it's very good at looking up one key at a time very quickly as long as it has the partition key so there's some kinds of joins we can do then we can do much more efficiently than actually doing a sort merge-join or even a broadcast join because we don't have to fully scan the cassandra table we can instead take those partition key values and pull them out directly so we added something called direct join and this is a feature that we've done quite recently and to actually accomplish this to actually make that new relationship what we did is we made a new physical node a new planning node because like we talked about before the the issue we we want to change here is not the what needs to happen but we want to change how it happens so now instead of there just being an option to do a sort merge join and a broadcast join we added another option called a DSC join so the DSC join is going to be very similar it's going to contain a join condition it's going to contain which columns it needs to read out how it needs to read them out and it's going to have children the children will be the source of the keys that it's using to pull things out of Cassandra so instead of the plan we saw before we end up with a plan that looks like this where we don't have a shuffle because we don't need to do a shuffle we know how to pull those keys directly out of the database and we only have one branch beneath it so basically whenever you're looking at a spark plan and you see two branches coming out of something that usually means you're about to do a shuffle or some kind of joint so this ends up being much more efficient I'm running a little low on time so I'm going to quickly go how we do this basically we take the plan we had before this logical plan remember this is before we go into that physical planning stage we look for which of the sides of this are smaller we check the sizes that are estimated to come out of both sides of this branch we look for one that's really small we check whether or not it is significantly smaller than the other side because we don't want to join if for example our database has a million keys in it and you're looking up a hundred million values then it makes much more sense to just read the entire thing before checking but if you're only looking up ten thousand keys out of a million values in the database then it makes much more sense to do this kind of direct look just look up 10,000 values rather rather than reading every single one so we look for the smaller side we take that branch and we kind of smush it together we take all of the nodes that existed on that path in the logical plan and we combine them into a single note so we basically extract the information out of all of those and make the des corresponding direct join physical node now like I said before one of the challenges here is that we don't know how to deal with the rest of it we don't know how to deal with what happens before this joint happens and we also don't know what happens after this joint happens so you might be doing this joint operation and then a whole lot of other things with the results we're not doing anything with that so we wrap that both parts in a plan later so once we've done all that we end up with the final plan that I showed you before we have this nice little tight direct joint here and we have our other table that we're doing a full scan on to pull out of Cassandra and we get the same results we would have gotten if we had done a sort merge joint so just to show some motivation as to why this was really important this drastically improves our joint speed so this is on a database that has 4.5 billion rows in it and these are all in the hundred million so just you don't have to count the zeros it's 100 200 300 400 million but if we're looking up one of these smaller subsets of keys out of the total number of rows we can actually do that much faster orders of magnitudes faster than if we just scan to both sides and then did a sort merge join the yellow line here is this time for the entire sparc merge joint where the red line here is the time required just to read the entire table out of Cassandra so we end up saving a ton of time by optimizing this join and we can do it all automatically so that's basically what I wanted to talk about today I'd be glad to answer any questions you have on adding rules to catalysts or how catalysts works thanks for coming and of course if you would like to know more about what we're doing you should join data stacks where we're doing all kinds of fun things like this all right yeah questions less efficient so one thing that is is quite particular between the two api's right now is the order of joints so by default spark does joins in the order in which you request them the cost point based join optimizer is a pretty late feature it's experimental you have to turn it on explicitly and do calculations of statistics before so that might be the difference is that if you say that there are certain joins it will not reorder those joints it will do the joins in the order that you have given them because that's one of those things where to actually do a good reordering it needs statistics unlike the cardinality of the keys and the total amount of data and if it doesn't have that information it's just like well whatever you gave me is what I'm gonna do so that's usually the difference underline both methods the Scala dot API and the spark SQL just the SQL string API they use the exact same optimization engine so there should be a 1 for 1 matching between every SQL string you can write and every Scala API the problem is you do get situations like that where parsing a string could represent a slightly different order of operations that may screw up the join order or something like that so that that would be my guess it also sometimes has issues with casting because you've written everything as a string in spark SQL it doesn't know if your literals are the correct type so usually if you haven't written them out explicitly assumes every literal is a string and then cast the other column to a string before it does a comparison so that's something to watch as well so you had one more question then we can go and it was on distinct I I don't know about that behavior in height I know in inspark the difference between like account with a column and account all by itself is whether or not those columns are gonna get passed through all the way to the source so for example you asked for something that requires no columns it will actually end up at the relation and tell the relationship you don't actually have to count and it read any data out of the database you know you all you need to do is tell me how many rows there are so that can actually switch up performance significantly we've seen we actually had an issue before we're in Cassandra if you did a select count sum column versus select count all of the columns you would get our select count versus no columns you would end up with a very different execution because selecting one column would prune it down to requesting only a single column out of Cassandra but we hadn't programmed it that if you asked for no columns it should get no data it should just return the row count so we ended up adding that as a later optimization so now if you ask for no columns it actually asks for no columns where before no columns was the same as all columns and one column was then much better than no columns so little things like that yeah so to register new data source type so there's two different things so if you want to add new notes this is a little more complicated this is where there are a few extension points if you want to do it you can do it live at the spark session so sparks session has inside of its Builder apply extensions which lets you pass in a lambda that given your given your session state I believe you can pass in more functions or you can pass in more plan nodes or analysis rules all that sort of thing what we use since we're selling the software to folks who don't really want to get into the nitty-gritty of it we want to preload all of those functions and extensions and to our our distribution there's a spark I'm going to forget the exact parameter name but it's something along the lines of spark SQL dot extensions which lets you provide a class that will be loaded by a reflection which is allowed to add functions and analysis nodes and planar nodes and things like that so that's separate than data sources if you want to add a new data source it's actually a lot easier data sources follow a different API that doesn't require knowing about all of this stuff you basically have to just describe given a a read how do you get that data out given these column requests how do you get those columns you don't actually have to think about the expression trees and how it's gonna fit in the optimizer it does all that for you so for that you just make your class and then a user has to have it on the class path and refers to it by package name [Music] mm-hmm yeah so I I know it's hard to see cuz I Adam all listed in very tiny text but basically it adds them to the bottom if you do cogent it will end up doing a map an OL RDD map partition with the generated code so in our case we we are not doing code gen in this implementation yet we're just using that do execute method so it actually takes an internal row and then doesn't mat partition but it also still just doesn't a partition so right now we make use of the size estimates table and Cassandra so Cassandra gives a very rough estimation of how much data is inside of it it's not very accurate but that's basically all we have at the moment we're working on trying to get some cardinality and other information embedded into SS tables so we can do better but yeah so since this code like I showed you basically lets us do whatever we want inside of it so we basically put inside of the actual planning rules the planning rules make a connection to Cassandra look at the underlying table that it's checking and pulls out that estimate data we don't push it into the spark catalog any other questions I think we have like four minutes yeah you keep running these optimizations every time you watch it is there anything you've experimented with remove certain tools that you know so within the spark optimizer unfortunately it can't tell whether or not they were really important while it's running so that's not something that's part of the optimizing right now it's actually pretty far away from a cost-based optimizer which is what you would need to do you basically need to collect statistics every time something runs and make sure those are stored somewhere the best thing that it can do at the moment is that if you run the same query more than once it does cache a lot of that so it'll be like oh you're requesting to do this thing that you just did well we actually know how that optimization played out it's not changed so so yeah sure there with difficulty I mean what you have to do is you just have to keep up every time spark changes internals we have to change with them we have to keep up I mean this is this is pretty typical is so one of the things that's currently in development within SPARC is the data source excuse me the data source v2 API so basically the data source API that easier way of incorporating your readers and writers into SPARC is being revamped because one of the big changes with allowing continuous stream processing where you have that per tupple base processing is that the methods that you're expose you expose in the data source API are really not sufficient because you have to suddenly deal with the fact that a record is going to go through the system not in a batch but one at a time so there's a whole new whole new API that's being developed it's being worked on right now it's not quite finished but it allows for a lot more specifics on that sort of thing like I've you've just gotten ten rows right then rather than you've just gotten an RDD full of rows right all of those another cool thing that excuse me that's coming in data source v2 is that currently there is no way of a data source exposing the internal clustering of the data that's being read so in Cassandra's case things that share the same partition key are on the same physical machine so if you do a read of data then a group based on that partition key we are already guaranteed that our partitions our spark partitions all contain they're basically organized by key already and since they're organized by key already we shouldn't have to do another shuffle before we do our our reduce or our group by so that's one of the cool things in data starts v2 is we can actually expose that into catalyst so catalyst can make smarter decisions about do you really need to do a co grouping before you do this operation so within the Cassandra connector so I didn't go over this in this talk but this is the basically the way we do it is if you imagine Cassandra has a token range and you can think of that as a long line that lists all of the possible data that's in the database we cut that into sections based on the ranges that are actually replicated within the database so this happens every time you do a full table scan so it'll basically say this range belongs on this node this range belongs on this node and then it will cut those up into pieces that match the expected size that you want of how much data you want in every spark partition and then attempts to execute those on executors that have the same ip address as the replicas for that token range so basically if you're looking for tokens 1 through 10 and you know tokens 1 through 10 are on nodes 1 2 & 3 it looks for executors that are 1 2 & 3 and attempts to run the task that reads that particular token range on those notes yep well I think that's my time I was told there's no runner so I'm I have to shut it down so thanks a lot [Applause]