funconf 2013, Matei Zaharia: Spark: Big Data Analytics Made Fast and Easy
Recording: funconf 2013, Matei Zaharia: Spark: Big Data Analytics Made Fast and Easy
okay folks so it's it's eleven o'clock so uh let's get started uh so thanks for coming to the talk um i'm mateysa herrera and i'll be talking about spark just a bit of uh i guess background on me if you don't know me so i'm i'm currently a a phd student at uc berkeley i've been doing research on big data systems for the past six years before working on spark i actually worked extensively on hadoop i'm a hadoop committer and i worked on the job scheduling features in hadoop and then you know a few years ago i started the spark project and after uh finishing at berkeley i'm going to be continuing uh to do research as an assistant professor at mit and so i'm excited actually to have eventually two universities involved in this project and you know more people from the east coast as well so let me just talk very briefly about what spark is uh how many people have actually seen it before or used it before yeah okay so these are number of people yeah so so spark is a fast and expressive cluster computing framework that's designed to be a generalization of of mapping use but it's also designed to be highly compatible with apache hadoop and it tries to improve two things efficiency and usability so to to improve efficiency it it offers uh in-memory computing capabilities which can make it a lot faster for applications that needs to you know reuse data or share data across computations and it also offers more general computation graphs and operators which lets you express complicated applications more efficiently to improve usability it has these rich language integrated apis in java scala and python so in scala it looks like you're working with scala collections and it also has an interactive shell you can use it from you know within the scala shell so what what exactly you know difference does this make so in terms of the the efficiency we've seen on wheel applications as much as 100x faster and you know another interesting thing this 100x is within memory data but even with on disk data we can go 2 to 10 times faster than just mapreduce because we have the more general engine in terms of the usability you can often have you know factor five less code especially if you're using scala which you know hopefully people here are um a bit about the the history of the project so it started as a research project in 2009 and we open sourced it in 2010 but since that point and especially in the past couple of years you know it's it's actually grown significantly as just an open source project and we at berkeley are running the project uh you know from from this open source point of view you know we're still doing research on specific aspects but we have a big team dedicated to just making it you know better uh more you know more stable easier to use and so on so today the the open source community includes 17 different companies that are contributing code to the project this is in the past year and we we have major users including yahoo intel adobe and you know multiple uh startups in the bay area and so so these these guys are contributing quite a bit and to further uh sort of help grow the open source community we've also uh entered the apache incubator uh a month ago um and apart from you know just just being the the core engine and the open source project spark is also increasingly an expanding stack of of higher level programming frameworks on top of it so that's actually one of the things i'll talk about today as one of our long-term directions for the project so on top of the core engine we've implemented a sequel this is the shark project which takes apache hive and and makes it run on spark we've implemented stream processing and spark streaming these are both released today and two things that are coming later this summer or this fall are graphics which is a framework for graph processing and ml base for machine learning and one of the cool things with implementing these on top of spark is you know not not only are they fast but also you can combine these things in the same application so they all have these nice functional scala apis and you can do like say sql query and then do a graph computation on the result and stuff like that or do it all interactively from the scala shell so this is part of a bigger you know stack of projects at berkeley that's called the berkeley data analytics stack or badass that's you know i didn't come up with that but but that's what it's called um okay so what will i do in this stock so the stock has a part at the a small part at the beginning is just kind of giving background on the project for people who haven't seen it but you know since a bunch of people have i also wanted to focus on some other things so after the background on spark itself i'm going to talk about two of the projects we're building on top and especially the way they integrate with scala and with the rest of the api uh these are graphics and shark um and then i'm going to talk a bit about the open source community this is you know one of the things that's really grown in the past year and there's some cool stuff going on and finally i'll talk about like what what we think is the most exciting part of the project which is the power to actually compose these different programming models and and uh sort of uh analytics tasks within the same engine and i'll i'll talk about you know interesting things we've seen in the engine using that and definitely feel free to ask questions throughout so let me just start with uh with sort of the the introduction so basically why did we want to make a new programming model instead of just say taking mapper juice and adding a scala api um essentially what we found working early on with with mapreduce users is that you know it really it made analytics a lot simpler but as soon as you got started with mapreduce and put a bunch of data into it there were a bunch of applications you wanted to do that it wasn't good for and these were first more complex applications than just one passive map adduce so a lot of say machine learning or graph algorithms have to do multiple passes or multiple operations over the data a second thing is more interactive queries so you've loaded you know 10 20 terabytes of data into a cluster you can compute a report on them and say two hours every night but now if you have a new question can you answer that question in in two seconds or do you have to wait two hours to to crunch through the whole thing um and then the final thing people wanted to do is is more real-time processing so you know you're building your reports every night that's that's awesome it's crunching a lot of data now can you update the same report in real time so something we we observed is that even though these applications look fairly different they actually all need one common thing from the engine which is more efficient support for data sharing across parallel computations so these are all applications where you need to share data across say time steps or iterations and in map address you don't have that all you have is like one pass of computation and then write your stuff to hdfs so when you try to do these applications with mapreduce you get stuff that looks like this so you know just as an example if you have an iterative algorithm like page rank you would maybe start with the data in in your hdfs and then do one step of mapping up here so i guess you can't see this yeah up here which is this blue box and do one pass of your computation and then to share the data with the next step you have to write it all back out to hdfs again and then on the next iteration of your page rank you load it back in and and you continue doing that or if you want to do ad hoc ques often these squares are you narrow in on the same subset of data but if that data set is in hdfs you know you have to incur the cost of reading it from disk every time and this is using using hdfs for this is slow due to data application each time you write it has to go across the network due to object serialization and disk io and actually when we looked at sort of real hadoop applications often you find for these kinds of applications that they're spending like 90 of their time doing this stuff as opposed to actually running the user's code so in spark you can do this sharing in memory between different steps and you can both create longer chains than just one map reduce or just ask multiple queries on the same in memory data sets so you can just load the data once and and share it very efficiently and this can easily be 10 to 100 times faster than the network or the disk so the way you do this stuff in spark is that instead of just thinking in terms of mapping reduced functions you think in terms of distributed data sets so you get this abstraction called resilient distributed data sets or rdds and these are just parallel collections of you know scala objects if you're using scala or python objects or java ones that are split up across the cluster and you can control for each data set whether you want to keep it in memory or on disk um you manipulate these data sets to different parallel operators we have a whole big array of operators i'll talk about it a bit and they're also resilient in that they they can automatically be recovered on failure so you don't have to worry about uh nodes crashing in the middle of your computation and to do uh to program these you use this functional interface in scala java or python so this is kind of what spark code looks like out here in this example this will be code you can type into the scala shell where you you have say a bunch of error messages in a log file across the cluster and you want to load them into memory and ask some queries so we're going to have a cluster here we have a master node and a bunch of workers and we can type in code like this so for example we can say okay let's have a uh a file a text file in hdfs which is our first uh data set and and we can represent this as a collection of lines so you just give spark an hdfs path and it knows how to load that and then this is our base data set and on that we just do transformations using scala functions so for example you can filter and just pass it or closure like to select out strings that start with error and you get back a transform data set which is just the stuff that passes the filter um so what the the thing we do under the cover here is we can actually take the scala closure object this isn't a dsl or something like that for strings it's actually just the scholar function you passed in and we serialize that function and all the variables it depends on and we send that to the cluster so it's much like working with collections locally and then you can do other transformations maybe for example you had these these error messages and the tab separated fields and the actual messages field number two so you can pull that out with a map and then you can also choose which data sets is an iron memory so you can say okay cache only the error messages in memory so what happens is as you write these transformations spark sort of remembers the graph of them and it actually will evaluate them lazily so at this point it hasn't actually run anything yet and then when you run another type of operation called an action it will actually launch a job to compute these so here we're going to say okay let's let's count how many of the error messages contain foo and count is called an action count has to give back a number to the program it can't sort of defer the computation any further so when we get this we come up with a plan to actually execute the whole thing and so what will happen here is that spark looks at where the data is on on disk on each node it sends tasks to process the data locally you know these guys do their work they send back results and the nodes also build a cache of of just the error messages they read along the way so that's you know first time you run this this is what happens and second time you run it spark will know that that the data is in the cache and it will hit the cache and and give you back the result um quite a bit faster so so this is kind of how it works you can type this stuff from the scala shell as well from standalone programs just to give you a sense of what you can do with this so one of the the demos i do is just loading wikipedia which is about 60 gigabytes on a 20 node cluster and just doing full text search of it just kind of stupidly you know find articles that contain a particular string and if you do this with on disk data with hadoop or or spark on disk it takes about 20 seconds to scan the whole thing from disk if you do it within memory data it takes about half a second so you can really get interactive response time on on fairly large data sets um with just arbitrary scalar queries and you know this also scales up nicely so we we can also do for example we tried a terabyte of text data on a 100 node cluster and you can scan that in about five seconds so that's that's kind of how you program it now the other aspect of the model that's cool is the way it it does fault tolerance so when you run spark you're going to use different operators and you're going to build this graph of data sets and dependencies between them this example at the top is you know a slightly oh actually i guess this is a python one because i stole it from a python talk but hopefully people will understand but this is a slightly um you know more complicated job where we we get a bunch of records um and then we we want to count how many records have each type so we map each one to the record.type comma one and then we use reduce by key which is our distributed reduce to add up all the ones for each key and now we get a data set of you know key value pairs for for each record type uh the number of such records and then at the end of this we're going to do a filter let's say we want to pick out only uh records that appeared at least 10 times so when when you do this so spark will build up this dependency graph it can go to sort of any number of levels and this is what it uses for fault recovery as well so if it ever loses some nodes in the cluster and some of your memory and a piece of this graph goes missing spark will go back and recompute stuff according to uh to the parent transformations so and it will do this for sort of arbitrary graphs so this is this is you know one one of the reasons that it's it's fast is the the data that we hide in memory we don't have to replicate it and you don't have to keep two copies of it because we know how to recover it this way okay um just to show you one one final example of the core spark i'll show you uh one of these iterative algorithms as well this is a simple kind of machine learning um algorithm called logistic regression so the the goal on this one is this is something you use for example for classifying spam you you have a bunch of labeled data points like maybe you have spam and non-spam emails you represent them as vectors and you want to find a line that separates them and that that you can then use when a new email comes in to decide is it spam or non-spam so the way a lot of these algorithms work is is gradient descent essentially you start with a random line and then you run these small steps these these parallel functions that will improve the separation between the two uh the two sets so what happens um yeah what happens on each step is you run this uh this gradient function which is just the sum over all the points that you do some math that tells you essentially how well the line is you know is classifying those points and when you do that that gives you a direction to move the line and to improve the separation and then you launch the same parallel sum again it gives you another direction you know you move the line a little bit and um as you keep repeating this process eventually your line converges to a good um separation so this is exactly example you know the kind of the kind of example i was talking about earlier where your algorithm has to do multiple you know parallel sums these are like multiple mapreduce jobs and they're over the same data so if you could keep the data in memory you could you could make this go quite a bit faster and so this is this is how you'd write this this algorithm in spark it's again this is like actual scala code you could type in and it's it's a it's a fairly simple uh piece of code so essentially there are two pieces here first at the top we start with say a text file and we parse it into points maybe you write like a read point function and we cache those points in memory and then at the bottom we have some iterations and on each iteration we do a map over the points to compute that that gradient function and we sum up the values with the reduce uh one of the cool things you know if you're using scalar is you can replace that line at the top instead of being a distributed data set it could just be an array or a list in your main program and the bottom part of it would still work you know so it's really like like working with your current collections so so that's that's kind of how it works and then this one if you look at the performance you get you know very similar kind of speed ups to what i showed before so if you write this kind of application in in say hadoop map reduce you know you get something that will run in parallel but it takes the same time for each iteration here's it's about two minutes because it's reading um the data from disk um if you write this in spark you get first iteration happens in about 80 seconds this is just loading the data into memory once and then further iterations are about one second and one of the you know really interesting things about this is when you write the hadoop version basically it will scale linearly it will behave really well it's it's a nicely uh you know behaved map ages job it basically is just mapping some stuff but because of the the nature of the algorithm um the the actual math you're doing for for each point is so cheap that that you're spending you know more than 90 of your time just dealing with the i o system in hadoop so this is the kind of thing where you know if you you unless you actually look at profile your job and see where it's spending time you could be hunting very and efficiently due to the hadoop engine okay so that's you know that's kind of a quick tour of of what spark does apart from these map and reduce type things that i showed and filter there are quite a few other operators as well we actually we we've tried to do you know similarly to projects like pig we've tried to provide a lot of the high level operators you might want just as scholar functions and we're adding more and more of these over time so we want to make it very easy to work with data you don't have to write your own you know sql joins or group buys or things like that and the engine itself supports any combination of these operators into a graph you don't have to do just map and reduce and it has a few other things that that make it faster as well it has for example support for hash based reduce instead of the sort based one that hadoop does which is faster if you're just doing grouping and joining and it also has support for controlling the partitioning of data across the nodes which matters it can it can help reduce the communication in a lot of these algorithms if you group like say you know web pages from the same domain on the same machine or stuff like that um so that that's the graph here is showing um if you implement page rank you know you can get sort of a 3x speed up just by doing it in memory but then you can get an additional sort of cx speed up by also controlling the partitioning of your data okay so that's you know that's that's kind of a quick tour of spark um now the the i think the the rest of the talk what i want to cover is what what we're building on top and and sort of where we see this going next so definitely so this is going to be uh newer stuff but definitely let me know if you know if at any point you have questions about this or about the previous stuff so i'm going to start with graphics which is a project that you know we've started building for graph computation so let's just do a bit of background on that so graphs i think i've done here knows are are very important um in data mining just some examples of of things you might do with graphs are if you have a social network for example you might use them to identify the influential people on it and then you know suggest them to others as people to follow or things like that you can use them to find communities find groups of your users that you know have some some common property or that maybe you need to put in touch with each other you can use them to target ads when a new user joins uh your site maybe you don't know anything about that user but as soon as they start adding a few friends you can start inferring stuff about them and actually start giving them more more meaningful recommendations and ads and you can also just use them to model complex dependencies if you're doing some kind of scientific or you know just even machine learning kind of workload so because of this widespread use of graphs there's been a lot of graph processing systems proposed in the past few years that try to do this on clusters and try to scale linearly and things like that um so just some examples uh google spargle was you know one of the first ones people sort of talked about google or the research paper about it and said hey this is how we're doing graphs giraffe is a open source implementation of peggle from apache graphlab is a academic and now also commercial project from university of washington and cmu which is a fast uh basically a graph processing engine for machine learning um and twitter also has something called uh i believe casa veri i think that's that's what this uh is a picture of but we couldn't find any logo for it so we just put that on there so so this you know people have been building a bunch of these and what these engines process uh provide are sort of two things uh one is an api to actually capture the dependencies between the computation so instead of thinking in terms of map functions and reduce functions in these systems you usually think in terms of being a verdict and saying hey you know i'd like on each step of the computation to see information from all of my neighbors and like somehow the system is going to give you that at runtime and find a good schedule to execute all these vertices and actually do do them in parallel so that's one of the things they give you and the second thing they give you is they try to exploit the structure of the graph to reduce the amount of computation and communication so examples of that are they might partition the graph so that you know vertices that have lots of edges within between each other on the same machine so you don't have to send that stuff across the network or they might say that hey each vertex is actually trying to do an associative operation like a sum of the stuff around them and uh they're going to to do that locally on each machine and then just sum up the results kind of like a combiner in map address so these different systems peggle and graphlab and so on basically have different apis for doing this so given this stuff the question i guess is how how is graphics different from them how is it different from graph lab or giraffe or pergol and um the the real answer is you know we're going to do a lot of the same stuff underneath but the thing we're trying to do better is uh simplicity so it the the main thing that that puts graphics apart is that instead of being a separate engine to do graph stuff with its own data format and its own runtime it's actually a library within spark and you can use that you can build graphs using the standard spark operations you can run these models like peggle and graph lab on them and you can also create the results of them using the same spark operations so what this means is that there's no sort of special work required to do etl in the graphics so to extract transform and load the data from whatever format you have in hdfs to whatever the graph processing system supports it's uh it's very easy to basically similarly there's no need to do anything special to consume the output of it the output of graphics is just an rdd it's a collection of scala objects and you can run your filters and counts and things like that exactly as i showed before because it's running on spark it also automatically gets fault tolerance and the fault tolerance composes across these things you do below and after it's like if you build your graph with a map function you know from reading from a file and then you do page rank on it and something fails spark knows how to recover both the map function and and the graph stuff on top um and finally you can use the spark shell you know similar to the the examples i was showing before for creating text files to query your graphs so once you build your your you know interesting kind of graph in memory or maybe as you're tweaking parameters to your algorithm you can just ask questions from the shell and i think this is a thing that that none of the the other systems actually have so um the the the second aspect that uh you know that helps here is um basically we the the for for sorry for programmability we also um offer a very simple scala uh apis with the same kind of ideas as sparks so working with graphics is actually really similar to working with different uh with distributed collections and on top of this api we've also implemented the graph flag graph lab and peggle apis they're each only about 20 lines of code so it's very easy to to do things in these models if this is how you want to think of your problem and you can also write calculations directly so page rank for example you can write it in about five lines of code so it makes it very compact so here's uh you know just a bit more detail on how graphics works the the thing we add to spark in it is something called resilient distributed graphs which is an extension of the distributed data sets i called before called rdds and these are a graph is basically a an rdd of vertices and an rdd of edges and we we actually store them in a special way to you know to take advantage of the partitioning and stuff but that's kind of how you think about it you just give it a bunch of edges and a bunch of vertices and graphics will organize them to make it possible to do graph computations and on top of this graph you know you can do your standard things like map and filter and so on and you also get three extra primitives that implement kind of the low level operations in pregal and graph lab-like systems and that you can use to express graph parallel computation and using this you can capture you know many graph algorithms just directly or if you want you can implement you know peggle and graph lab on top and get that api so let's just see some code example of how you work with this um so here's here's what we're going to start with so this first part here is just spark code we're going to load in a graph and we have a text file in hdfs which is maybe web pages maybe these are vertices and then we have a second file which is edges like the links between uh different pages and we just load these as you know spark text files and for for the the edges we're going to do a map to turn them into an an edge object maybe it's like tab separated the source and destination so this is how you make your graph you just run some spark code then you say okay new graph and you pass in the vertices which can have any data type and the edges and you can also ask to cache it in memory okay so as soon as you have that you can just start asking some questions about the graph like let's count the number of vertices in it that's just going to count that rdd or let's count the number of edges so you can do this kind of question sort of interactively and you can also transform the graph so say you want to get a sub graph maybe you want only the uh the the vertices in this thing that uh you know reference berkeley somehow so you can just pull out the pages that that actually contain berkeley maybe you know it's one of the fields in your graph and that just gives you a new graph object and finally you can run graph operations on them in graphics we're actually going to provide a standard library with a lot of the common algorithms so you don't have to write pagerank again even though everyone says like hey you know you can write page rank and three lines of code in our system you know really you want to write it in one you want to call an optimized implementation so so you can do that and when you run just to see what happens with the output then so when you run pagerank you get back a new graph object which has a extra information on its verdicts it has the rank um and you can so so you get this graph of ranks and now you can do stuff like say sum up the the ranks of all the vertices or you know print out a specific one or save this thing to a text file and just to show you that you know if if you like implementing page rank it's not too bad this is the the code for for the page rank as well so you know there's a bit of stuff going on there but it's basically this little thing here it's using the paragle model built on top of of graphics and it's just giving it some functions like basically a a function to apply to each vertex and then a function to combine the values across the vertices so this is this is still a little bit in flux but basically the final api will look something like this so that's that's kind of a quick tour of graphics um so basically to summarize we have this distributed graph abstraction on top of it we wrote the pagle and graph graph lab apis or you can just use it directly and on top of that we're actually writing a lot of the the common algorithms you might want so you don't have to do those yourselves and we're still working actually on optimizing this we have some some some pretty interesting things that we're doing but but the early performance of graphics is also pretty good so just this example here i'm comparing our page rank on a 16 node cluster with uh with hadoop and we're you know about i guess eight or nine times faster on this one already and the reason we're faster is the combination of the features i mentioned before so definitely in-memory caching helps hash-based operators help because we don't care about sorting this data and also controlling the data partitioning so not having you know when you're going to join data sets a bunch of times partitioning them so so they can all be joined locally on each machine and this is still you know under construction but we think an alpha release of this will be ready this fall so you'll see it when it's ready basically we just want to make sure we optimize and tune the api to something that we can support later on okay so that's that's one of the projects um the second project i want to just talk briefly about is shock as i think we're building on top and then i'll talk about how we're actually putting these things together so graphics is you know is more for programmers who want to do graph computations shark is uh if you just want to quickly ask sql queries um so shark is is a project that's actually been out for i think a little bit over a year now it's a column oriented uh sql uh analytics engine that's built on top of spark and it's it supports sql it's actually basically the shark code base is apache hive which we modified to run on top of the spark engine so it supports the hive dialect of sql it also supports complex analytics because it has a scalar apis so that you can call into it from from scala and and combine your sequel with with scala code and it tries to be highly compatible with apache hive and basically it supports the existing hive language user defined functions uh serializers and deserializers and scripts and it can run on unmodified hive warehouses you can just launch it and point it to your meta store and start asking questions uh so this is because we we base the the the code base on the sql part of the code base on on hive itself and it's been it's actually starting to be used at a number of companies uh the one that's been doing the most with that so far is yahoo actually where a team has been both helping us build shark and using it for uh for in-memory uh analytics on their ads data so um shark's performance uh you know is similar to what you see in sparks so uh this is uh some um some results on sort of real quiz that uh one company that is using them on this kind of two terabyte data warehouse and basically you can see we're about five to ten times faster than high for on disk data and we can be you know 50 to 100 times faster within memory data and this is again for you know similar reasons to to the other stuff before so it performs well actually we've also tried comparing that with things like like cloudera impala that i you know c plus plus and generate a whole bunch of code in llvm and we actually perform very similarly to those um but the the part that is you know that is kind of cool if if you're a scala user and that i think actually really sets this apart from the the other fast sequel things people are doing in this space is that it also integrates very nicely with spark so using shark you can just run you know hive hive quays through jdbc or through the shell but you also have this api to run them from scholar so what you can do is in a scholar program you can write your more complex analytics functions like your logistic regression or the graph stuff i showed before the page rank and then you can do sql to rdd so here's a sql query run it and give me a distributed data set from it and this is you know similar to how in jdbc you get back a bunch of records here you get back a bunch of rows that are distributed across the cluster and then you can just pass that into your your machine learning functions or the other functions you want so what will happen here is that shark will actually run in the same sort of spark application as as this it will share the same caches you know the same in memory space and it will build this result in memory and then just feed it directly into the scholar function so it's very cheap to move data back and forth between sql and scholar and to write an application that combines these two things um and this is we're actually we're actively working on making this better as well so one of the other things that will be coming up is being able to hunt scala and then save that result as a shark table and and query it with sql as well but this is this is one of the cool things you can do um and i don't have time to talk too much about the other projects but there are other things in the stack that are neat as well so if you're interested in stream processing uh spark streaming which has actually was released earlier this year is a api to do that and it lets you take all these functional operators you get in spark and apply them on streams along with new operators for windowing so you can do fault tolerant streaming computation and then if you're interested in machine learning something that's available on github now is the beginning of our of the ml based project which is this library called ml lib of high quality implementations of machine learning algorithms so this is a thing that's on github and it will be in the in the uh spark 0.8 release which will hopefully be sometime in august and it's basically you know it's it's a team of machine learning researchers at berkeley that are working on this okay so in the last uh part of the talk i just want to talk about sort of where the project is heading and things we've seen and i wanted to talk about both open source community and also this you know this idea of composition that that i think is is is one of the coolest things about the project and that's where a lot of the research in spark is going to be so let's start with the community so we we've been you know very um uh fortunate since we started the project to have a very rapidly growing community and i just want to tell people a bit about it because i think if you if you saw a spark talk like you know a year ago or two years ago all these numbers would be many times smaller than they are today so just some quick stats on the community we have over a thousand meet up members now in our san francisco bay here meetup i think again last year this might have been like 500. we have over 60 external developers who have contributed the spark over its lifetime these are external to berkeley within berkeley we have maybe 10 to 15 people that are working on it and we have 17 companies that are that have contributed in the past year and again i think if i were giving this stock uh a year ago this would have been maybe five or six so it's really uh you know it's really been a blast in in terms of uh growing community um this past year in particular has been very transitional so um it i think a year ago basically uh more than the majority of our developers where if i'm inside berkeley today most of them are from are from outside so we have i think 40 people who've contributed in the last year um we have uh many of the patches that have come in now in in in addition to being sort of small fixes or improvements we're actually getting large sort of multi thousand line patches from outside which were things that you know otherwise it would have taken us weeks or months to build ourselves and also most code going into the project if you count by lines of code is now from outside uc berkeley and we're doing you know a lot of what what i do for example is just code reviewing stuff and testing it making sure it's something that we want in the project so just some examples of things that were contributed from outside in the past year this is just the selection there's been a lot more stuff but these are some of the bigger sort of multi thousand line patches team at yahoo contributed all of our support for apache yarn and so this is i think two or three thousand lines of code and it's not just adding hooks to launch spark on yarn but it's also adding things like hack locality uh awareness in the scheduler to make it on well on sort of multi node clusters and so they they did a quite a bit of refactoring of the scheduler to provide that um a team at intel in in asia has actually been rewriting a shuffle implementation the shuffle is the network transfer between in sort of distributed reduced operations and this is one of the most performance intensive things you get you have to do because a lot of applications are going to be network bound it's also it can be quite memory intensive so you have to be really careful with that it can do a lot of io operations as well and they've been working on improving this and you know they've given us a bunch of new patches that that make this quite a bit faster another team at intel implemented a fair scheduler for jobs in the same instance of spark this is actually a thing i had implemented in hadoop so it's kind of ironic that we never built it in spark until someone else gave it to us from outside and the teams at yahoo that are looking at shark are providing some pretty cool features there as well so like one of them is is compiling uh sql queries into java byte code instead of just interpreting the sql which is what hive does another thing is actually column oriented uh compression um and then we've got an other stuff from the community that's more about integration and build but that's also like pretty you know pretty time consuming to do ourselves so for example all of our support for scala 210 has actually been contributed by this company called imagina which did some really non-trivial things like hacking the scala interpreter in 210 to make it work with uh with with spark clusters which is a thing we've had to do separately for each scholar release and same thing with a lot of the packaging in the project that's been contributed from outside by companies like clear story data so we're really excited to have these things and i think as we've been ramping up we're also getting better and better experience in terms of of you know actually accepting these and and testing them and so on so we we're going to continue doing that and essentially we we are working closely with these contributors we're happy to work with any new contributors as well you know we're getting more and more experience doing this and the other thing we're focusing on at the lab itself is making it easier to evaluate these kinds of changes so you'll see probably in the next few months sort of much more thorough performance test suites and regression suites that are going to come out to make it easy to evaluate changes to spark so if you want to contribute on this stuff you can you can join us on github so let me um so let me end with with one final thing which is sort of where where spark is going and uh things we've learned along the way and i think the really cool thing is here is this this aspect of composition so basically you know you've seen all these slides about spark you might be saying sort of well it's it's great that it's it's fast it's easy to program because of the scala api but what really makes the project unique you know when when there are so many uh open source you know computing projects happening in the big data space what what actually makes spark unique so the thing that we're doing uniquely and that i think is going to matter a lot is this idea of unification so we on spark we don't just provide batch processing or just map reduce we provide multiple programming models as diverse as sql or streaming or graph processing on the same computing engine and so if you look at the stack you know we have these projects that are built on top of it many of these were essentially separate runtime systems in in most other stacks but here they run on the same engine and they can all share data efficiently in memory through the spark api and this idea of unification has two powerful benefits it has benefits for the engine itself for actually building a good engine and for the users so from the engine perspective uh something we did recently that's kind of fun to do is let's look at spark and compare it with the leading open source frameworks in these different types of analytics application you know streaming sql all that stuff and we'll compare them along two dimensions code size and performance so when you look at code size this is kind of what you get this is looking at we have hadoop mappers on the left you know hadoop is is a big project i only took the map radio spark but it's still a pretty big thing we have cloudera's impala for sequel we have storm for stream processing and we have giraffe for graph processing and you know these are each 60 to sort of a hundred thousand lines of code spark today is about 25 000. now some of this is because we're using scala so and you know which is actually a good thing you you at least if you're a scholar developer uh you can sort of appreciate that building these things in scala makes them a lot easier to maintain but it's still you know from this point of view uh it's actually it's doing a lot with a small engine but then the part that's that's really cool about this is if you add in um so these are four different engines if you add in the components of our stack that do the other programming models so if you add in if you look at sql for example you know impala is this whole separate engine you have to run besides hadoop map pages that's going to do fast sql i sql implementation shark is about 12 000 lines of code just on top of the the core engine if you look at stream processing storm is again this whole separate code base does a lot of you know special stuff for fault tolerance and scheduling our streaming implementation is about 7000 lines and if you look at graph processing you've got this whole giraffe framework and graphics itself is is about 2 000 lines so this is a really powerful thing is that we with a small you know fairly small team of people at berkeley and you know growing but but still kind of early on open source community we're able to actually capture all these models you know in in in the same engine with quite a bit less less repetition of work and then this is the code size if you look at performance in all these models were actually quite competitive so this is a benchmark we did comparing sql performance of shark against impala as well as amazon headshift and this is showing both on disk data and then memory data uh with the head shift unfortunately you can't really control whether the data is on memory or on disk it's basically going to cache it in memory after some time and you know this is just one of the queries but in in these cases so we're competitive with impala on disk and we're actually faster than than these things in memory so you know we're we're doing fine i mean all these projects are going to improve over time but it's not like we're a factor of 10 behind or something um this is comparing our uh streaming performance this is a throughput graph so actually higher is better and when we compared against storm this was you know basically a while back but it was sometime last year we were going about two times faster than storm so again we're in the same ballpark and this is the graph performance if you compare the page rank with uh with giraffe and graphics we're again at kind of a similar speed so this stuff you know it's going to change over time and it depends on the benchmark and so on so you can definitely take it with a grain of salt but what i'm trying to show here is that we actually have competitive performance in all these types of workloads using this much smaller and more integrated code base and the final thing that's cool about the the performance aspect is that whenever we do optimizations for one of these projects we're actually helping all of them at once so this has happened to us numerous times as we were developing the spark stack just a recent example when we put out spark streaming we did a bunch of optimizations to make the spark scheduler work well for short sort of sub second map edu straps because in spark streaming we wanted to do many small jobs and incrementally add in new data and then when we were writing a paper you know on shark after that to compare shark sql performance as soon as we switched to the branch that had these optimizations all the short queries were going about two times faster so this is a pretty cool thing that you know you're not going to get if you improve things like storm and impala independently so this is kind of the performance perspective and from the user perspective i've talked about this a bunch already but hopefully you realize there are some some pretty cool benefits as well so one of them is that applications can easily compose these models all these models have the same kind of scala api and this abstraction of distributed collections and so it's very easy to say harness sql query then run a graph algorithm on it or run a machine learning algorithm on the result and just combine those in one program the composition is not just doable but it's efficient so the data stays in memory across these steps the fault tolerance works across them you don't need to checkpoint the data or like save it to a weird data format in hdfs to load it back in into the next processing step so that's efficient and also the models get this this interactive shell that we built for the original project sort of for fee so if you want to analyze your you know your graph data interactively ask some queries you can just create that graph in the spark shell or actually a cool thing you can do in spark streaming is if you want to interactively explore the state of your stream the state of it is just a bunch of rdds and you can ask these queries on them exactly the same way you would for you know for for some other in-memory table and so this is i think it's a cool aspect that you know came out of the way we built the project and it's also one of the things that will be our main long-term direction uh in terms of research and there are a bunch of cool things you you can do with it so apart from just you know taking the existing models one of the things we're looking at is what other programming models and types of computations we could build on top of the the same engine and we're looking at adding new primitives to the engine like say incremental computation or mutable state that will let you express more types of computations efficiently um so so that's one of the things the other thing uh we you know we we want to look at is uh optimizing better across these different programming models so for example in scala 210 when you have macros you can start peeking into the functions people are giving you and you can do stuff like push down predicates if if you're going to do a sql query and then a scholar function on the result you can push down some filters from scala into sql and start optimizing across all of these and so we think that you know in the long term if you look at big data analytics many projects today are just focusing on getting it done like how do i even do machine learning or graph computation but as soon as people have tools for that to really be productive they'll need to be able to combine them and this is one of the things that you know we want to support here so that's that's pretty much what i wanted to cover um so maybe one one more important thing which is how do you get started if you want to uh to play with spark uh we we have quite a few resources online for getting started and you should just go to sparkproject.org to check them out some of the cool things available are video tutorials screencasts that will let you set it up and use it as well as hands-on exercises we have a set of exercises on ec2 where you launch this little four node cluster it comes populated with some data and you can use all these projects to step through to analyze this data and it's easy to run spark either locally on your laptop uh or on different types of standalone clusters including apache mesos and yarn or an ec2 you can launch a cluster that we reconfigure for you and then if you want to learn more about it in person we have a training cam coming up at the end of august in berkeley this is a two-day event where you can use spark you can do both hands-on exercises and just see sort of tutorials with us doing them and you can do it in person so i invite you to check that out so to conclude big data analytics is going to evolve is going to con combine more complex analytics more interactive and more real time and in the spark project we're building a platform that unifies these it's open source and we invite you to try it out yeah yeah um on the topic of unification i'm curious whether there's any ongoing research about heterogeneous genius compute infrastructures oh yeah in particular is there an active collaboration with the light with delight yeah so that's a good question yeah we we've actually talked with the light group at stanford quite a bit and they have a few projects where well they've been taking similar types of computations and trying to map them to run on top of gpus and things like that so i actually want to collaborate more with them to try to actually run some of these on top of spark yeah i've talked with them a bunch of times yep yeah the data um some of them are you know you should send me an email so i think they're published in separate papers but if you want them all together i can probably try to get you the source for each one yeah definitely yeah so the benchmark so at least the the sql benchmark is it's actually uh uh i think you can hone yourself we have a page about it and we have amazon machine images you can run it on so all all of that is designed so you can launch it yourself and we want to do the same with the other ones yeah yeah let's say you're sorry uh oh let's say you lose a node yeah so yeah so when a node gets lost essentially um we lose some some partitions of an rdd and we he can start just those missing partitions and actually one of the the nice things that happens there is like usually a node has not just one partition but maybe like 20 or 100 so we actually spread those out across the rest of the cluster and we compute them in parallel yeah it's similar to like when a node fails in map reduce and you lose like the map function output yeah so we we do partial recompute yeah any other questions yeah so one way to view uh spark i suppose is that it provides this orthogonal set of abstractions on top of which you can build all these things yeah yeah was that the goal from the beginning or yeah no so that's a really good question so it it um it really wasn't i mean it's a thing that this is just me being honest like otherwise i would have said oh yeah we knew we had a more general thing but but uh it really um it's a thing we discovered as we went along so when we started spark we we were only really focusing on machine learning computations and then new things like peggle came out and we looked at them and we were thinking like oh no could those also do this stuff well and then we realized we could um we could implement them so basically at some point after that we switched to a mode where we were more actively seeking out other things we can do um and and also seeking out abstractions like common abstractions we can add like say control over the data partitioning that would help improve a bunch of these together um so yeah so it didn't start that way but it's um in a sense also because of necessity if we wanted to do a graph thing or a sql thing the first thing we did was try to do it on spark and then if that worked we didn't have to build a new engine for it so that's kind of how it came out yeah yeah so uh in your example yeah yeah good question yeah yeah good question so now it doesn't automatically try to figure this out uh this is the thing we might add in the future it's uh well i think doing it well is is pretty hard but um but that also makes it interesting at least from a research point of view probably the first place where you might see that is in shock where it might decide based on how often you access tables and stuff it might automatically decide how to cache them but we don't have that yet yeah any other questions yeah no so cache invalidation you also have to do yourself we don't try to propagate updates to the graph that would again be a cool thing to do but this is just to keep it simple for now yeah yeah currently yes yeah when is the transition to 210 yeah good question so we so i think the the spark 0.8 release which is going to be uh you know soon hopefully in the next month that will still be using two nine and after that we're going to switch to 210. so the reason why we pushed it a little bit is just because there's a lot of new stuff that came in into 0.8 like a lot of these you know changing the shuffle and stuff like that or or the scheduler and we don't want to make too many big changes at once but there is a scala 210 bench that's you know pretty close to being in sync with with the master branch so you can use that um and we you know we hope to like when we release 0.8 we're also going to update that branch to have you know kind of a scala 210 version of spark 0.8 yeah yeah okay any more questions or all right so i think thanks for coming and it's time for lunch you