BDSBTB 2015: Spiro Michaylov, Developing Spark SQL Integration for MongoDB
Recording: BDSBTB 2015: Spiro Michaylov, Developing Spark SQL Integration for MongoDB
so good morning everyone I want to begin with a short caveat I work the tableau software but this isn't a tableau talk so you really don't want to draw any conclusions about tableaus interest in Scala from this or about you won't learn anything about tableau support support the sparkle MongoDB or relationships and partnerships with other companies that might be represented here and you know tableau tends to integrate with everything so those relationships do exist and I don't do this work for tableau software so one thing I will say though is if any of you have a secret desire to build query processing infrastructure in C++ just quietly grab my attention in the corridors afterwards and we'll go hide our shame and talk it an undisclosed location because we are hiring but at any rate I I really do love scholar but it's something I do in my spare time okay so I'm going to tell you about a personal journey overall and then focus in on a particular part of that journey about a year ago I decided I wanted to learn what it took to integrate apache spark with a database system that it didn't lure it wasn't already integrated with and I kind of poked around a little bit to decide what to integrate with and I was curious about MongoDB I noticed that it's becoming increasingly important particularly in the last year or two I think of it as almost the epitome of a non-relational database and collections have no a priori schema so I saw it as a particularly difficult integration in some ways or a particularly challenging one so I make no value judgments about MongoDB I think it's kind of interesting and I think it stretches us as a community and that's really the attraction for me now I used to advise graduate students many years ago in another life and sometimes people would come to me with things that were kind of an awful project to get started on something with and in retrospect I think maybe this was an awful starter project because there was so much to learn but I had a lot of fun and maybe find some of it interesting and maybe you can benefit from some of my experiences and also one of the interesting things that happened is that why after I got started on this the external data source API of spark sequel was introduced and I got particularly interested in that again if you work at tableau your kind of interested in integration with other data systems or easy ways to integrate so there was just a curiosity that was already in place there so this focus this talk focuses on using that API and maybe helping other people to use the API there are lots of other aspects of what I've had to deal with in the past year that are really interesting and I'm not going to talk about them things about how to take advantage of MongoDB partitioning and then map it to partitioning in spark how to infer a schema from a MongoDB collection managing connections to the database particularly in a partition setting type mapping between MongoDB types and spark types actually generating the DB query these are all interesting things but I really won't talk about them my agenda I'm going to give an overview of the state of the art as I perceived it when I got started on this project and I'll talk briefly about sparks JSON support because that was what kind of got me interested and it gave me the impression that maybe I could succeed at integration with MongoDB it seemed kind of analogous then to basic rdd level integration and then move on to spark sequel and kind of how its API fits together and then introduced the external data source API and how to use it and how it integrates how it interacts with performance issues which were really that I think the plot that I find most fascinating and then give you an overview of how it all turned out for mongodb or has turned out so far and some of the lessons that I've learned okay so the state of the art as I saw it then was I saw that Spock did support JSON you could read a JSON file and spark would deal with it reasonably actually infer a schema for it as I'll describe in a moment and you could you could make complex references and queries down into nested documents elements of arrays and so on get complex results back get a raise back get documents within documents back in sparks equal and it dealt with nails reasonably so if you tried to delve it down into a document that didn't exist in certain records in certain manga in certain JSON documents it treated it as an also reference through a null was it now so it was all quite tidy and I now know that perhaps that's not that surprising because of sparks desire to be reasonably compatible with Hadoop and with HBase I guess i'm not sure if those were the actual motivations but i was a little surprised and impress them so then the where did I get my ideas about how to actually achieve this I was by no means a spark expert and I'm still not a spark expert i looked at the existing MongoDB connector for hadoop some people actually use that through spark I don't know if they've actually gotten around to a native connector I looked at the existing open source Cassandra connector for spark which is actually pretty advanced and it's actually where a lot of people start to learn this kind of stuff and then I didn't know about Calliope at the time and actually I just heard about it in a talk yesterday and I actually keen to learn a little bit about it okay so that's where I moved that that's where I started reading code and learning and just so I'll just give you a teaser for a moment what I noticed at that time with sparks JSON integration is that you could you could you could present it a simple document and it would infer a reasonable schema for that document right I'm using some features I'll actually introduced later if you're not familiar with them for showing what schema sparkers inferred and for a single document it's fine if then you might you might presumably you have more than one document in your JSON collection if these documents aren't in conflict then the schema is kind of just the obvious merger of the schemers in the individual documents the problem with j with inferring a schema for json of courses that the schemers of the documents can be in conflict who can be an integer in one document and and a nest of document in another document so what do you do spark takes a very pragmatic approach to that is that it really just starts to approximate the scheme and starts throwing in strings so there's some loss of functionality there but you can actually keep querying and you can keep you can keep delving into deep structure and you can keep getting complex fragments of JSON documents back so it all it I viewed it as it works reasonably well it makes reasonable compromises and I moved on from there ok so the way one tends to begin to integrate a data source with spark is to is to create a custom rdd to implement the abstract class ided so the most people who are looking to figure out how to do this look at the cassandra connector i think it's on github i might be confused about that and see how it's done there and now that's a that's a rather sophisticated implementation so the then so the abstract class rdd is really the spark secret source from the point of view of the integrator and what you have to do there is provide a way to describe the set of partitions and then how to get the data for each partition right assuming that you're going to want the data partitioned which you eventually will for good performance and then spark uses your implementation in the usual idd lazy way when it needs to when it needs to get the data for individual partitions then you actually also have to implement the traitor partition from sparks point of view that's just an ID it's jeff see that the partitions have ids 1 2 3 etc and then anything you need to squirrel away in your implementation to know how to get a partition what the partition represents maybe it just represents a certain key range information that you've queried out of MongoDB metadata or whatever database you're using you can store in your implementation of partitions and you'll have access to that when you need to a tip that I would bring up and what a group at a number of places here I've got tips on a how to get started easily now we won't necessarily get good performance just by following these tips and then also about where I think some of the pitfalls are so a tip is get started with that nothing is forcing you except for performance considerations nothing is for forcing YouTube to partition your MongoDB collections in this case or anything else you can do everything in one collection I'm sorry in one partition and then once you've got that working move on from there and that you know partitioning can be a very difficult and involves topic if you're if you're motivated to get data out of your back end as quickly as you possibly can okay so what does this look like encode just just a sketch so you need a class that extends our DD and it's parametrized with whatever record type you're going to define for for for the for the data you're returning and I happen to use just the aleve actually with a just a wrapper for the Java bindings for for the MongoDB query results and then so partitions is the array of your partition implementations and then compute is have spark tells you which partitions it wants you to return an iterator on and there are there are lots of complexities you'll find out implementing this properly you have to do resource management probably in the construction time for this class you'll have to go and get some meta data out of the database you'll have to pass in what table or collection or whatever you need to connect to and then in the compute method you'll have to make sure you do some cleanup and they get it outside the scope of this talk but clean up can be a little bit tricky you want to make sure you're not leaking connections to your database and so on and a question up to the front information of the partitions within your spark types and I'm wondering if I mean you it sounds as though you sought to gain performance by doing so at any point did you regret having put awareness of partition mongodb partition information into your your classes here in spark I haven't yet but on the other hand the partitioning work I've done is not that sophisticated right I I do observe mongodb sharding I don't necessarily observe it well for example i'm at one mongodb shod to a partition which probably isn't the right thing to do in general it's just not something I've delved into anywhere near as deeply as say the Cassandra people have and they've really gone to town with that issue looking at what they've done if you're really serious about that then look at what they've done okay so moving on to sparks equal zooming in so spike sickle provides a way to when way or another register our dd's as database tables and then query them using sequel using your existing sequel knowledge essentially it's arguably the least mature feature of spark but it is very powerful and the APA's where I think marked experimental until spark 130 when they were stabilized those kind of an upheaval right at that point of stabilization that was when the dated frame class that you've heard about in a couple of talks already perhaps today and yesterday was introduced and I'll say something about that and you know to illustrate how powerful this can be you can even use it to join tables from completely different data sources although as I'll get to that's perhaps not something that's going to be very efficient in the near future and spark although other people may know better than I do when when that'll become efficient so the thing I found about sparks equal in general is that it deals with surprisingly well with non relational data and more smoothly and naturally than say Apache drill and I don't mean to be unkind to Apache drill I think it's very powerful but I would describe the Apache drill view of non relational data as if your data isn't really Julia you haven't hit it with your baseball bat enough times right you just you keep massaging the data within your drill query to make it relational enough for a drill to deal with it on the other hand you know mapper has gotten involved with drill Ted awnings for cats day today seattle back in june actual has suggested to me that maybe drill is going to be moving in the sparks equal direction just as a point of interest so just a very simple sparks equal example here without using an external data source registering a collection of objects of some case class and then creating registering them as a temp temporary table and then doing a very simple select star query from that temporary table then we get we get back a data frame on which we can call print schema and that's where we got this lovely representation of the inferred schema and then we can print out the data frame in a tidy format that i won't bother to show you here so there's a very trivial example what are the components of sparks equal river though for our purposes there are really important ones our data frame which is the you can think of it as actually the the data that's being queried but also the results of a query or also a data frame and they contain a schema that is what is the schema of the data in that data frame and they also contain an RDD of rows and a row object is really basically for our purposes and array it's a lot like the the forgotten what the results are individual result objects in jdbc are but it's the same kind of idea its API is kind of similar so that this was the subject of the upheaval in spark spark sequel in 130 it used to be that the relevant class was a schema idd and it extended idd of row the data frame contains an RDD of row and both of them container schema and there's a there's a nifty dsl for the data science people who used to our and that sort of that's not relevant to this discussion and then so schemas are you basically the schema type in Spock sequel is called strap type these types all live in or misspelled apache spark sequel types and you just see a very simple example of here of how to put together a type that's just a row of of they in this case and integer and a string and then you can see how to construct the corresponding row object that corresponds to these schema so in general these rows do not are not self-describing you if you traverse a row wanted to operate on it in some way you kind of have to traverse the row in raw in lockstep with the with the struct type to know what's actually you know to know what the even to know what the names of the fields are in general so now we come to the kind of the the core of the talk sparks equal external data sources so they introduced the ability that in sparks equal you can use ddl you know the the data definition language of sequel to register to refresh to list and dearest register external objects as if they were tables among other things okay and so then these objects can be queried use xbox equal DML the general query language and in recent versions those object these this data can actually repair the the registration can be persisted in the hive meta store and can also banned the the objects can be updated so this is a little unclear in the slide it's thieves that is a definition of these external objects that specifically can be persisted so you specify the connector type by naming its class and then you specify the actual object and the connection properties by passing a property bag in that's eventually passed to a constructor we're in your implementation and so that what a key thing is that the API allows the addition of these data sources without rebuilding spark you just drop the implementation in well not a classpath this being spark and being distributed it's in a few class paths and that can be fiddly and this is in contrast with drill where unless this has changed very recently you can dynamically add additional instances of a pre-established external resource type but if you want to add a resource type you actually have to add code and rebuild drilled as my understanding and there there may be some trickery that's not well-publicized there so as I said this was introduced in spark 120 it's been evolving since then and one of the wonderful things that happened in the 120 releases that the pre of the pre existing data sources that were supported were actually reimplemented in terms of this API which was a great thing because that API wasn't at all documented at that point so that was the only way you could find it find out how to use oh and those example of those data sources actually produced provided pretty nice examples this this model is I think really beautifully designed for ease of adoption as I'll show it's perhaps not designed for optimal query execution and I would imagine that there'll be some development over time so that you have the option as an implementer of an external data source or an integrator of an external data source to opt for easy implementation or really efficient execution or hopefully first the easy implementation and then to incrementally improve it and there are elements of that already in the design the sparks equal ddl that I was talking about so really the relevant part here is create temporary table you give it a name if you don't give it a schema the schema is expected to be in third and you say what the qualified class name is of your implementation for this connector and then you pass some options you can also specify a select to actually populate the thing and then you can describe and refresh table and so on for this talk I'm focusing on registering existing data as a temporary table and inferring the schema and then being able to query from it I've experimented with some of the newer features like updating and being able to create the table from a select this way but I haven't really delved very deeply into that yeah so the API components are in the package org apache spark sequel sources an interesting thing that's happened as these have changed and extended is that there's a little bit less inheritance between the various traits and so on and i guess that was a philosophical thing but there are really three kinds of reusable things here traits and abstract classes and such one is the basically the providers which are there these are these are what you have to implement to have something you can register with sparks equal then the relations are just essentially two really the way is the way the API holds on to what you've produced and then the scans are your description of how you're going to pull data from the external data source right base essentially what happens here is that the spark query compiler search box box sequel query compiler makes table scan calls on your API and you're going to then give it the table day return the table data that at once and there are lots of nuances here that I'll get into now so what's the overall the recipe for creating an external data source you have to implement the provider class and this is cold when you register a table you have to implement one or more versions of create relation then it returns your table scan class as a base relation that's how the API deals with that and then you need to end it it also implements the trait relation provider scheme a relation provider if you're going to have the user specify the schema it's a regular relation provider if not scheme relation provider if the users got to provide the schema and you're not going to infer it and then creatables provider if it's actually if spark is going to pass in a data frame for you to send out and specific the external data source then you need to implement a table scan class this implements the basis of relation abstract class and then you need two choose a build scan overload depending on which of the table scan traits you implement we're going to get to that in a moment this is called when you when somebody actually queries it can optionally implement insertable relation if you if you actually want to support updating okay we'll see some outlines and then some examples as well so relation provider again you you need to implement a create relation so just in plain relation provider it takes a sequence sequel context and that property bag that the user handed over when they registered the table for a schema relation provider you'll notice that there's also a scheme at the end of the argument list for passing that in and then for the create a bull relation provider you have to specify what happens in the case of a conflict and then otherwise the the difference is that in addition to hear the the sequel context safe mode parameters you also have a data frame here's the data to go and persist okay base relation and insertable relations so base relation is really a lot of the substance here this is where you have to provide to spark sequel the herbs box equal uses this to find out the schema that you've inferred for this data or the schema that it had previously passed in and then you optionally can override the size in bytes this is a hint to the cost base cost-based optimizer to give it some idea of how large the table is for purposes of joint optimization and so on I haven't actually experimented with this yes but they could be important for performance in certain cases and then the trait in suitable relation is if you actually want to support updating as I said so for performance they're really the two important things to think about our filter and projection push down if you just implement table scan then that means that that the spark sequel optimizer is just go to us you give me this table the whole table please give me that table the whole table please you'll have to return all of it even if the query just needs one piece of data and so then Spock sequel will do all the work that's great that's a great way to get started so that's my tip there is to actually do this first but from a performance point of view obviously this is going to be hopeless so what a filter and projection push down projection push down is when you you make it possible for the optimizer to say I'm only interested in these columns could you just give me these columns although there's a caveat here and then filter push down is well the query had some filters and I'm going to massage the filters so that they're relevant to your table because the query might have involved several tables the optimizer figures out which here are the filters would you apply these please use the underlying database to execute those filters efficiently send back only the relevant columns and only the relevant rose and then spark sequel has to do very little work now the intriguing thing about sparks equal though is that it actually reapplies the ownership I should stop doing that it actually reapplies the filters after you've passed the data back it doesn't one way you can view it is it doesn't trust you to apply the filters correctly I think a more charitable and positive way and actually realistic way is that it doesn't force you to apply the filters so it'll give you it'll give you this bag of different kinds of filters resulting from the query maybe some of them I'll talk about reasons to ignore them but you might have multiple reasons for ignoring some or all of those filters it also may choose not to give you all of the filters it may keep some for itself to execute just because of the way the way the query plan that it infers works but again the more filters your reply will be more efficient you know spark reapplying the filters will actually be pretty cheap if you return a little data then let's see the real of the main thing and I'll hop on this a little more later you need to make sure that your query you isn't too restrictive you don't want to fail to return data needed for the correct answer of the query that's really important so filters very easy projection is a little bit more finicky if you implement the version of weather table scan where it expects you to project you actually need to return just the columns that spark sequel asks for which may be a surprising set of columns it may have a surprising order and it may I think I've even seen it for the columns in a different order from the order in which you specified it in the schema so there are there are lots of case classes defining the various kinds of filters and they've dribbler being dribbled out in various releases I guess I put an asterisk on 150 because unless it's happened in the last couple of days it's not out yet but equal null safe seems to be slated for introduction there and I've highlighted or in the in the one point three point 0 introductions because it's a little bit dangerous so why would you ignore filters will notice that they're being dribbled out when you know a few / release and this will you know this is a healthy project there'll be lots of innovation over time so when you when you build a connector you you might be worried about well what if new filters are introduced well you can just hopefully you can just ignore the new filters that are being introduced until it suits you so either they didn't exist when you wrote your code there's no corresponding filter in the system in the back end you're using or you think maybe it's inefficient there it's not worth pushing through or you ran out of time to implement it or below just you're just busy there's a tip about these filters these were all straight forward until and or and not were introduced if you if you get a filter that is the orbiter disjunction of two things into some reason you can't handle you can't translate one part of that disjunction you better not drop it right because now you've got to start returning to a few records rather than too many so in these in these and or and not combinations you need to make sure that if there's a sub filter you can't implement you replace it with true effectively by massaging the whole expression tree or just putting in a filter that corresponds to true in in that part okay I predict that's going to generate a few bugs in a few implementations so then what are the scan traits so this table scan you just return a row you get told nothing prints can you just hold the requires columns note that they're strings prune filtered scanned you get the columns and the filters catalyst scan is something I haven't delved into but it's intended to be to achieve a greater performance and I think we'll see this develop over time simple examples of what what kind of behavior you can expect so a very simple select on the left hand side catalyst will request the column a and will give you the filter a greater than 5 to optionally deal with so this is my test whether i put the audience to sleep in this second example i'm saying that the requested columns are going to be a and B but the query is select a so anybody why is catalyst requesting both columns a and B in this case it's select a from T where a greater than B anybody that's correct i think the the the suggestion was that that the spike sequel will reapply the filters after after the data comes back and it can't apply the filter a greater than B unless it has access to both a and B so this is actually an interesting consequence of this real ease of adoption approach that in that it's another case where there's a loss of efficiency because extra data has to be you might have a very good implementation of filtering in your connector but you still have to pass back the extra data because unless you use the I presume unless you use the catalyst version of the catalyst rate you act sure you are required there's no way to stop a smart sequel from reapplying the filter a caveat that I think is kind of important for mongodb once you get into complex expressions everything works fine but the uppers catalyst never asks you for these complex expressions it actually has no way to ask you and I'll finish up in just a moment here it has no way to ask you because it's just passing you a string so it just doesn't it doesn't push down filters on them at all and it just asks you for the top level column for for the info schema and I'm going to have to move forward a little bit just an example of generating a query fairly basic is this powerful enough for mongodb yes and no yes it's powerful enough for the expressivity maybe not in that you'll get awful performance because of what I just showed you about complex expressions if it happens to be important to your query and also it's not arguably it's not practical for aggregation right group by clauses and so on because there's no way to push those down to the back end so I'll pass by my hookup example slides and then so the summary that this all started out simple but poorly documented it's become much more complex somewhat more complex but better documented and it provides you the option of a really simple integration to get started and I think that might be a winning idea although there might be some things to tweak and contrast this with a patchy drill where you really kind of have bear the full brunt of the complexity of the optimizer are quite a lot of it to integrate with it I applied this to what's probably the most problematic DBMS sort of the court of the graph databases right and even then it wasn't too bad I think I have some serious concerns about performance for deep document references and aggregation and it could have been better easier if the API dealt with with complex references and finally I'll say I don't really understand the performance characteristics in practice it's really quite seems decent for easy examples and we know that there are cases where it could be deeply problematic there are lots of other interesting technical issues i'll pass over those and for more information my blog a river of bytes kind of Chronicles my learning experiences along the way this is all available open source don't try to put it in production please it's really more it you deserve what you get really are and maybe I do too but that's another story and there are some bad w you can find out about this from a number of other sources thanks very much listening so we're running like a little behind so um how about maybe if you guys have questions go ahead and ask him but other than that wolves you