scale.bythebay.io: Flavio Brasil, Spark Datasets: why they aren't great and what could be done
you okay thank you so yeah thank you for attending my talk today when I'm name is Walker Brazil I work here Twitter I work on the car data library's team and we basically work on scalding sunny birds and a little of spark as well but today I'm going to talk about personal open source project that I've been working on so initially the title of the talk was spark data sets why they aren't great and what could be done about it but actually I decide to do something about it so the title of the talk now is clearly spark a type safe Scala API for spark Segal okay so I would like to start with a brief history of the spark api's and why the api's were introduced just to understand how this new API came to be so so are they so in 2014 the first version of spark was released and the first API was RDD so we most of you probably know already and this is an example that my involvement going to be using throughout the talk that is given a collection of tweets we turn the top hashtags okay so so first we flatmap get the facts out of the tweet object split into words and then we filter only hashtags and then we map to lowercase hashtags and this is not too intuitive but basically you create like tuples with the words and a counter right and then you reduce by key so the key is going to be the words the first elements of the table and we reduce the second value to the counter so we have a count of each word hashtag hashtag word in the group of tweets in the data set of tweets and then we call top and ordering by the second elements of the collection and we get the top hashtags okay so this is a nice API it's typesafe you can use color functions you can use color types and that's totally fine but it has some downsides so one of the problems is for instance the users can introduce inefficiencies so here we have a filter operation and if we move the filter operation to after the reduce by key operation it's the same meaning right we are going to have the counts for all words and then only filter the words that are hashtags but that's much much less efficient right because you are going to do a shuffle for all words and then only get the ones that are hashtags so that's not really efficient and that's a problem of the API for this case it's easy to see it's not efficient you move the filter but there are many other more subtle scenarios where we can the user can introduce an efficiencies oh that's really a problem for an API the second problem are the Scala functions so when you call flat map and you have a function that reads the text and split into words this is something that is just for this park execution it's just a pointer to some byte byte code it doesn't know anything about this function doesn't know that it uses the text column doesn't use any other column for instance or is going to split the text and this means that it limits the number of things that spark can do with this transformation so if tweets are coming from for instance a culinary format like parking it would be much more efficient to load only the text column and not load any other column and that's like orders of magnitude more efficient for instance but SPARC can do anything because it met it needs to have a full to it object just to call this opaque functions color function another problem is the API introduces as transformations that we return are DD and it also has actions so the painting on how which method you use you can do just a transformation that's going to be executed later or you can actually bring the data to memory so here the top operation is actually bringing the data set to memory and it's really common to see newcomers people learning or even people that are using for a while the Big Data solutions bring in data to memory without understanding the actual implications of this operation right so that's why that's not so ideal and then as a way to improve that and give spark a better way to optimize the execute of a job in 2015 thus far community started working on something that is actually all using something old that is sequel so with sequel if you see a sequel query you have all the information about what's going to happen right you have all columns that are going to be used you know about all transformations because it's just a strange that you can parse so that's something really the execution engine can introspect any aspects of this transformation so what and then a new API was introduced as a way to build this kind of sequel query that is datasets sorry data frames and data frames like are not so great but they have a very nice property the execution engine knows everything what that's happening for so for instance the explode split selection explodes more or less like a flat map for data frames and then you split so the execution engine knows that the only the text column is going to be used for instance and it can load only the text column from a corner format or it knows that it's going to explode or it splits and apply more optimizations huge for the execution and also you can move the filter so you could I could move the filter to after the aggregation and spark would be able to optimize the execution accordingly and that's something that's much better because you don't need to worry too much about subtle scenarios where you can introduce inefficiencies and in 2016 another API was introduced so the data frame is fully untyped you don't have like data frame of tweets you have just data frame and in 2016 a new API that these datasets was introduced and this is how you do the top hash tags using data sets and it's like a mix between RTD and data frame so we have some operations that are untyped data frame operations and you have operations that are data set types operations so if you go back back and forth between things that are types and things that are not typed and you can do interesting things because for instance like this elect is going to use a text column and the execution engines know knows about that you can also use the flat map as a scholar function regular opaque Scala function but there are some problems for instance let's say that instead of selecting the tabs and then doing a flat map I replace both operations with a flat flat map that reads the tags and splits in towards so it means that spark doesn't have the information about the columns that are going to be used anymore because that's an opaque Scala function make sense another problem is a type safety so if you reference a column you have to say the type of the column and that means that you can make mistakes and have real times errors and type safety is a really important thing for its color developers right so that's the problem basically like the data set API has like this in one hand we have type safe Scala functions that are not interactive all right so you can see this execution engine can see through these Scala functions but on the other side we have strings and columns that are not type safe but they are fully introspective all right so an ideal solution should be able to give the execution engine and having enough information about the transformation but at the same time preserve type safety so how could we do that so in 2015 I came across this paper it's a really interesting paper from fiddler and some other people I don't know if you know Philip Wadler but it's a very important person for the functional programming community he worked on Haskell or monads or free moment so he defined many things that we use today and he wrote this paper with some other people that is everything old is new again quoted domain-specific languages and it's a recipe for building a language interpreted query in any language so I decided okay that's an interesting interesting challenge so I'm going to implement that in Scala and I really use in 2015 the new open source project called quill so right now it's a mostly a database centric library so it's very much where you can talk to Cassandra talked poster great so many types of databases and we are at the 2.2 version we have a maintenance community and it's one of the most popular solutions for database access in scholar and now we have a new module that is as part integration to use this Park sequel engine I'm going to give an example just to show you how oil works so here we have a simple query so selecting from couples and then we're couple one person is older than the other and then selecting the name of the other person and the difference of age to the other person and it's a very simple query and that's something that can be expressed in Scala very similarly Marla's with the same structure that is using fire complications so let's say that couples is a collection in memory people with a collection in memory you can use for complications to apply some filtering get one person other person couples and then wield the result so the yield is the Select clause in sequel the if is the where clause and the body of the far comprehension are the from parts of the current query and but that's really nice like we are using Scala you're using typed transformations but that's something that we don't want to do we don't want to load everything from data from the database to just rather to run some queries we don't want to load all couples of people so what can we do there is a technique that is actually an old technique called quotation and the paper that I use to develop this library uses this technique as well and it's interesting because it's using the cold as data instead of cold that's going to be executed so we wrap this block of code with quote and what's happening here the block the quote method is a macro method and it's going to receive a parse tree from the Scala compiler and then transform this parse tree into an eternal AAS so you use this walk of codes instead of being something that's going to be executed at one time it becomes information for generating an eternal ASD and this is more or less the ASCs possible to read it here but it's a very simple iest has all the information so it can see through flat maps you can see everything that's happening in that computation right so and then what the paper defines is a set of normalization rules sorry it looks like super complex or her so hard to understand but this is something really really elegant if you see the same transformation intimated by slick in Scala or microsoft lync you it's like thousands of lines of code and this is like a couple hundred lines of code in his column and it's much more reliable because it's based on that theory and transformations that are principled so what wheel has is gap the ast using quotation pass through this normalization engine that has that is based on the paper and then from that from the Scala codes we can generate the same sequel statement so that's what quilt does or it obeys access this is a demo of coil because it does something more that is compile time quotations so you can do like repetition that's a Curie fetching people with named John and then plantation pay attention now when we print the query you can see the generated query at compile time so the quality SL that Coyle implements it's a compile time coated DSL and it is able to generate the final sequence statement at compile time as well so you have very nice feedback loop and you also have better performance and there is another thing that I do here in this demo I change the column H to H E and then the compilation fails that let's see here so that's the first demo like show you the query and then I change the column and the compilation fails because it knows the query at compile time so it can be configured to access the database and probe the query against the database so that's something that it's possible to do with quail at all so I've been working on this that is a version of quail with an integration for its Park it's very similar because Park has a sequel API so you can call spark context equal contacts sequel and has a Seco string right so I integrated quail with this API so this is the top hash tags using quill so we first we use this round methods that is also using quotation it's a it uses implicit quotation but this block of code is going to be quoted and then I can lift a data set into a coil query right and then this is where less likely flat map that we have for our disease but that flat this it's I call concat map I'm going to explain the hats later what's very similar and I can also like Francis move the filter and this that's going to be fine because the sequel engine is going to be able to move the filter back to the best position in terms of efficiency and also the API is totally lazy when you are within a quote block or around block everything that's there is going to be generated generating a sequel query so you don't need to worry about having something that's going to be called and bring things to memory like actions everything is lazy everything is easy to reason about so yeah so that's the the sequel that coil generates from that con that apes call api and as you I said before the spark Seco engine can apply so many optimizations because it knows every single thing that's going to happen in this computation right so as a last note quail has a memetic API I actually might background mostly back-end systems so I was working on the tweets micro service but I changed T teams and work started working with big data solutions and one thing that I really didn't at first was the flat map function that's calling or rdd's data frame data set they have it's a function it receives a function that receives that returns a traversable what's right so if normally flat map implements the bind operation of a monad in Scala right so I would expect something that like this but that's not the case because rdd's and data frame datasets calling typed pipe they are all applicative so you can't compose them as a moment but actually quail queries are walnuts so you can do flat map and return a query and that's going to be composed of better composable so just one example of composability here it's a query that that returns couples with the same age right so you we use a far comprehension use --lift query for one person another person couples and then join by the couples table and find people with the same age and that's like something that's much easier to do with coil than doing with rdd's or an applicative api so that's what I have like if you are interested please try it out give us feedback so we can improve the new integration I think it has a lot of potential but it depends on usage if we have users right and also I would like to say that my team here Twitter is hiring so if you're interested with working with me or scalding so many birds let me know as well feel free to to grab me like and talk after the talk and I think we have time for one question maybe so as Quayle was integrating with the data set API you can if you have a function that is not something that's going to be sequel you can use just the data set API so you can do a transformation using the sequel API and then you can do map or flat map using a scalar function so you don't need to embed any UDF for executing regular is color cold makes sense oh I see okay okay I see for quail you would oil doesn't do the integration with spark to like to find the UDF in spark you would have to do that by yourself but for calling arbitrary functions you can use something that same fix and quill that you can call any function or any sequel statement I guess that when there are some things that are inside at work right my question is how for my people have a combination of when you put something yeah that's a good question so anything that you do within a quotation needs to be something that can be translated to sequel or else you'll get a compilation error and the completion errors I can say that they are great right but at least you have this type of this safely of all only having things that can be translated to zero so oil doesn't have all the features that sequel has or spark has so you might hit some limitations within like integration between quill and spark sequel for Seco in general it has a pretty good support for most features but it's not complete like sequel for so many databases in sequel have so many differences that really hard that's really hard to to find like to implement all the features right so that's one of the limitations this Park sequel integration is very young so there aren't users I think right now so I just released so that's something to consider as well so weightless rate for people who know star and you know they don't have to learn the scale today the set of people who know sequined Domino's now is is Institute larger so for these people you give you telling them you get all these great features of we know but you need to learn stout have you faced this problem and so yeah that's a great question because we see that here at wither a lot like people that know sequel and they prefer to the sequel and I think that's all to be fine if it's someone once you do sequel like do sequel is the best way for them right but I think this the target of these libraries more Scala developers are going to do big data and one type safety and the features that Scala provides but that's a very good question we have time for yes okay so last question [Music] so quail for database access already has features for accessing most databases so and oil also has a superset superset of the features that sleek has so it's a replacement first week ok or an alternative trees lake okay okay so I think that's it thank you very much for cutting today [Applause]