Keynote IV: New Developments in Spark

BDSBTB 2015: Matei Zaharia, New Developments in Spark

Recording: BDSBTB 2015: Matei Zaharia, New Developments in Spark

thanks thanks Alexi yes thanks everyone so yes I'm going to talk a little bit today about new developments happening in spark and you know I'm actually kind of happy to give this talk because you know in the past at many conferences I went to they were a scholar conference or Hadoop conference and and I just gave the talk on what spark is but now since this is scholar in big data already you know I'm assuming that people have seen a little bit of sparkle I could actually talk about some of the things going on inside it so just so you know once lied about the company i'm at database so database is still a pretty new company it was founded by the team that created spark at UC berkeley and we remained the top contributor to the project we actually you know even though the many contributors to spark we still contribute more than half the code going into it and the last I checked we contribute about eight times more code than the next highest company in terms of spark so we really dedicated to going spark as a project and we commercialize it through this end-to-end hosted service on ec2 so if you saw you know all the activity going on with say connecting spark to visualization or notebooks or things like that this is a way to get basically all that in a pay-as-you-go fashion on ec2 but that's kind of enough about the company what is you know our goal and sort of our direction for spark so from pretty early on we wanted with spark to give a unified engine for all big data workloads and platforms by platforms I mean sort of one time scheduling systems or data sources or things like that and we thought that would be extremely useful because real world big data applications involve many different data sources many platforms that you have to connect and having to deal with you know tons of separate engines for each one is very complex and slow things down and you know doesn't doesn't let you be productive so this is kind of what we've been trying to build in spark so underneath spark you have the different sort of platforms at Quran on these are either one x or data sources and you have things like Hadoop but also many other things that are not Hadoop such as cloud storage systems and an environment like Amazon Web Services or OpenStack key value stores like Cassandra even traditional databases you can connect to them using spark and then on top of spark you have the different workload support at different types of applications streaming sequel machine learning graph batch jobs and so on and the nice thing about designing kind of the engine this way is any application you right on top you can then move across these different environments or even connect together data from two different environments very easily you access them in the same way and likewise if you're someone you know plugging in a storage system into spark suddenly you have access to all these applications on top so what's been happening in inspired development if if if we look at past two years in particular you know there's been a very fast growth of spark and much of the activity in spark has actually been in these libraries on top and integration points underneath and there's been a ton of progress made in the say so in the past two years we've got a whole new library for spark that became actually the most actively developed component which is spark sequel and its data frame API in different programming languages we have a very fast growth of the other library so i think the machine learning library goo'bye around a factor of 10 since two years ago and there's lots of cool algorithms being added in there we have standard pluggable api's for things like data sources and we also added the our language very recently as a new kind of set of tools that spark and integrate with and as a result of this we're also starting to see very diverse use of spark you know initially when it started people had one or two Apple occasions in mind like say interactive data analysis or machine learning those are really popular ones at the beginning and most of the use cases were on Hadoop but actually if you look at stats more recently we see spark being used quite quite a quite far beyond Hadoop as a storage or runtime system and also for quite a few different things so we we ran a survey of spark user is about a month ago and we see actually among the users we surveyed only forty percent are running spark on Hadoop yarn there's actually pretty much equally many that are running on public clouds for example and that's an example of this unified engine you can move across many different data sources and and you know gain value from all of them most users actually more than half of users use at least two of sparks built-in library so I'm talking about sequel machine learning streaming and so on which is also exciting to us because we want those libraries to connect very well together and among our customers we see you know of course tons of customers use Scala which is which is really great especially when they write full-fledged applications or batch jobs but also ninety-eight percent of them use sequel and sixty percent use Python you know often in combination with things like Scala and the reason for that is you have many people who are not software engineer spy merrily but who want to access data sets and they can do it through these interfaces but the other thing that I want to talk about is going beyond the libraries so there's been a lot of activity and libraries but one of the the nice things about the waste park is built is that all the libraries build on the same high level API and because the API is high level we can also make interesting changes underneath them and in the past two years and also in sort of the next year going forward we and others in the open source community are working on some of the biggest changes to the spark engine since the project began some of them already in spark some of them are still underway so these are the things I want to talk about and the nice things about these is that they do immediately affect all the libraries built on top as well so I'm going to talk about three things project tungsten which is a effort to improve CPU and memory efficiency network and disk i/o improvements which have mostly gone in in the past year and adaptive query execution which is a new project new feature that that we've started on you know basically this month and I'm going to start with with project tungsten so project tungsten of this this effort to look at CPU efficiency is really motivated by hardware trends and hardware trends basically you know Drive they certainly drive the whole you know highs of distributed computing in big data and they drive a lot of other things in the industry so if we look at hardware trends quite a few things have changed since since spark voice came out so spark first came out in 2010 and back then this is what hardware looked like so you had for storage you had a pretty much everywhere you had hard disks each of them could read at say 50 to 100 megabytes per second for Network you had one gigabit per second and for CPU you have you know sort of three gigahertz processor is maybe with multi-course and in this thing you know the thing to notice is the storage and network bandwidth are far more limited than what the CPU can pass is it you know Thea gigahertz you can easily process more than one gigabit or more than 50 megabytes per second so really all these distributed systems were about optimizing the use of these two resources getting data locality avoiding use of the network and so on if you look in 2015 things have changed a little bit so for storage of course there are still a lot of spinning disks but you also have new storage media that are a lot faster such as SSDs where you can easily get ten times the bandwidth of in terms of throughput of hard disk for network virtually everyone you know has moved to 10 Gigabit Ethernet but unfortunately for CPU is still had basically the same speed of course and not that many more chorus part you know for our server so basically if you look storage network have improved by 10 X but CPU you know it's still essentially the same thing so this project tungsten is about improving sparks CPU efficiency and basically preparing it for the next five years of hardware as well as applications and it's a consist of sort of three main pieces these apply sort of throughout spark off heap memory management is one thing one time code generation and cache of our algorithms that actually use the CPU cache efficiently so how does tungsten plug into the rest of spark they're basically a few different interfaces to it that that will all plug into it and the ones that it really touches right now are sparks equal and data frames although some of the stuff that's happening is also designed to improve just normal rdd api and more of the stuff we have planned in the future is going to you know to affect that and basically all these things map down to a commoner presentation of what they're trying to compute which is a data schema for the data and then a query plan with with you know various operators you want to put together and then given this stuff there are different backends that tungsten can hang on to start with we're looking at just very efficient execution within the JVM but we also you know looking at sort of generating say native code and is something like llvm or using things like GPUs or using more exotic hardware such as non-volatile ram that is coming out that's adding a new layer in the storage hierarchy so just before I dive into you know the things that are happening I want to talk a little bit about you know one of the ways you can use tungsten today except for sequel which is the data frame API so the data frame API is a pretty new API we added we added at the spark at the beginning of this year and it's based on this concept of data frames which is a very common way to work with tabular data in our and python and many data science environments and it lets you do bay basically relational algebra same things you do in a database like filtering and joining it lets you do math and statistics and it also lets you easily load data from many sources write it out in many formats and so on and data frames as a whole have become you know extremely popular programming tool this graph here is just showing the transfer them and basically a combination of our where this is the main way you work with data and Python where the pandas library let's use data frames have kind of driven this trend from sparks data frames give you the same kind of interface in all of sports programming languages very similar to these are single node data frames out there and basically you can think of a data frame as a collection of structured records so it's like an RDD except you know the schema of the data you know it has say five columns and this one is a string and that's an indoor something like that and the other thing that's different about them compared to the normal spark API is they capture many operations using this limited kind of expression language which is really a domain-specific language and what that means is that spark doesn't just see an opaque like Java function in there where it's like okay who knows what's going to execute what I on this it actually sees an expression like for instance you know user equals matei or age is bigger than 5 and by understanding these expressions it can actually come up with with more efficient ways to execute them so there's a little example of the data frame API in Scala you know basically you load for something like a JSON file and then you can write these expressions so this is filtering out things where the user field is my take and then you can do group by and you can do some and because of data frames understanding the expressions and generating code at one time it can be faster to use this then even than traditional rdds because it can actually fuse together many of the functions and things that you would add some overhead in normal execution so that's that's kind of the current you know the main way you would access our this stuff right now so how does thunk then help you know performance for these things so there's three things first of all is off heap memory management the idea here is the store data outside the Java heap and this is for two reasons one is to avoid the overhead associated with java objects it's nice to program with just java objects floating around it's you know very natural to plug in your own code and libraries but java objects add a lot of overhead per object and that begins that up when you work with big data sets and similarly garbage collection you know works fine if you have small programs or short-lived things but when you're working with large collections you know of objects it can it can really slow things down so the way we do this obviously we have to store these you know we have to decide on a binary format the store that meant if we put them outside the Java heap for'ard it easy we just use civilization libraries luckily there are many fast ones coming out for Scala that you are used a lot of cool features of scholar to speed these things up and for data frames and sequel we have a binary format that we can operate on directly using the second part the code generation so this is very similar to how you'd write something and see and just point to little the places in memory and act on them and what does this mean you know in terms of efficiency so we see a big space saving often it's at least a factor of two but sometimes it can be more especially for things like strings or nested objects that had a lot of overhead in Java and the second thing it enables is we can use these new storage devices now like flash and CDX points that look like memory sort of but you probably wouldn't want to run a JVM on them because you know they have higher latency and they're not exactly the same thing so that's kind of nice that you can plug these in and still keep other stuff in in the normal Java heap second thing is runtime code generation so especially once we have this binary format we need to actually have code run on it and the nice thing about the kind of expression API in data frames as we can understand the expressions and we can actually generate really efficient java code and we can even combine together many expressions that you're using in the same program and generate code for the final combination of them and this avoids a virtual function calls and it avoids the overhead with from from genetics and and boxing in Java and these things again add up when you're you know when you're trying to improve CPU efficiency and right now these things are both available for sequel and data frames as i said but actually there's lot of opportunities for code generation even in the koi API and machine learning and so on so some of the things we're looking at our code generating serializers for your data type fusing together functions or if you do many maps and filters it avoids a lot of virtual function calls if we can fuse them into one one operation and things like math expressions in ml lip so hopefully you'll see this used throughout the rest of spark soon as well and the final thing is cache of our algorithms so you know even though you can use memory it's pretty fast it's still you know basically it's still a bottleneck compared to how fast modern processors can can actually do computations and it helps a lot to minimize memory accesses and to use the CPU cache as well as possible so now that we can control the memory layout of things in you know many areas of spark quite a bit better we can also design data structures for cache efficiency and one example that's gone in recently is a prefix sort based on this this idea called alpha sort that existed before and the idea here is when you're sorting records usually you have an array of pointers and you you know you move the pointer is around then each time you compare two of them you follow the pointer is to compare those two objects but that leads to a lot of memory accesses so in this setup you can actually store a little prefix of the key with each pointer and then most of the times when your campaign to records the prefixes are different and you can just compare you know just those two without having to do extra memory accesses to pull out the actual records so this is one of the examples of a cat friendly algorithm that helps and you know how do all of these add up there's a graph with some sort of really early performance results in just one benchmark that compares a default spark with code generation so this is really comparing sparks equal basically first just the default the way it turns you know without any of these features then code generation on heap use of tungsten so this binary format and then a few pieces of tungsten and you can see it helps you know helps things go faster and scale out farther away as well by using these things so that's that's kind of project tungsten one thing i should say about this is most of this stuff i explained here will be on will be in spark 15 which is coming out in you know two or three weeks and will be on by default in it so you'll be able to try it out in that version some of it is experimental in spark one dot four already okay our second thing is networking disc so you know i talked about cpu efficiency a lot but there's quite a bit of stuff to be done with the network and disk as well and this actually is something that we focused on a lot in the past year there's not as much new stuff going on in this so basically network and search speeds improved by a factor of ten as i said but that speed you know isn't it doesn't just immediately affect all applications you actually have to work kind of hard to really leverage this and that's because there are quite a few challenges with keeping these devices busy all the time on disk and even on SSDs you want to do large operations doing many small ayos you know isn't isn't as efficient as doing large ones in the network you have to keep you know network connections busy and when you're on a cluster of machines there are many nodes and you also want the load balance them across them you don't want everyone for example although he just asked requesting data from the same map task at once because then you know it's not going to use the rest of the links in the network so there's a bunch of things you have to do to keep them balanced and finally you have to do all this from multiple cores that are all talking to the OS at the same time and you have to use many disks together on the same machine you know to actually use the full bandwidth so we we've spent the past year we actually we Haute bit by bit almost all of sparks networking layer to improve on these dimensions and you know the it actually does lead to significant improvements I don't have you know time to go into detail on all of these but the one thing I wanted to show is one of the results of this which may be people saw which is the sort benchmark so we decided to enter spark into this industry benchmark that's existed for a long time that was started by Jim Gray and it was started to measure basically advances in hardware and software he said okay let's you sorting as a standard task let's just pick you know one type of data to sort and let's compare different hardware and software types across many generations to see how things are improving so it's been running since 1987 and many of the entrants into this build hardware or hardware is less common these days but definitely software specifically to do sorting that's like all that the software does whereas in spark we just wanted sorting to be one application you right on top of it so last year we entered into the largest scale category of it which is called Daytona gray sort and here you have to sort 100 terabytes of data there hundred by two records in a fault-tolerant matter on a distributed system like say HDFS and we actually were able to set a new record for this using spark we tied with a team from UCSD that had built a system just for sorting that's all that the system did and we did it using two hundred nodes and we saturated eight SSDs full of bandwidth on each node and 10 gigabits of network so and these these are the things that were possible because of the improvements in the network layer and we also did it on the public cloud so it's actually the first time that a combination of public cloud and open source has won this in the past people use their own data centers and sometimes their own software so this is a comparison to the previous record so actually this is an interesting benchmark because you can bring whatever Hardware you want as many machines as you want if you have 10,000 machines you can throw them all at it and try to sort this hundred terabytes and the previous record kind of alternated between systems that were really efficient on a small number of machines and systems that were really scalable so the 2013 one was by her yahoo with Hadoop and they used two thousand machines and sorted this data in in 72 minutes and the machines were actually no pretty new machines they had I think around 2.5 gigabytes of bisection bandwidth per machine each machine had a 10 gigabit link they had lots of disks lots of i/o bandwidth as well but even though they were actually pretty reasonable with spark we were able to do this faster and also with significantly fewer machines because we were able to actually utilize each machine folium if you look at network bandwidth per machine we actually pretty much filled up the 10 gigabits that that were available here and the Hadoop one didn't use you know even a tenth of the actual network bandwidth available in the cluster and we also were able to scale this up the sword even better byte of data and about 10 times longer which was about 4 hours so there's a little graph you know what when this benchmark is running which shows the total network bandwidth across the cluster I don't know if you can see it but basically you know there were 2 207 machines and it's around 220 or to 10 gigabytes per second and this is this is basically one point one gigabyte per second per node which is around nine gigabits and that's actually pretty good if you're using TCP you know on a network that's pretty much as fast as you'll go there's two pieces here because we I think this is one of them is like shuffling the data while we're sorting it and then another one is writing the output to HDFS which also goes over the network for application that's right let's do two little pieces here ok so that's all and this stuff is already in spark even in the current released version and and the final thing I want to talk about is a new project called adaptive quay execution and this has to do with with gray planning as opposed to just what each machine does you know it's not just what each one is doing but also do you have a coherent plan above for what they should do so quay planning is actually one of the most critical things to performance in a distributed setting because even with the fast networks today it's still way slower to move stuff across the network than to keep it on one machine so things like the level of parallelism use and operators you know d spread data out to many machines or these few of them and they HD more stuff locally you know or to put it the other way setting the number of reduce tasks in MapReduce is very important also the choice of algorithms for example when you're doing a database join you have broadcast and shuffle join and it's hard to do quite planning well for big data workloads even with the techniques used in databases today like cost based optimization you know databases excel at quay planning but they do this because you have the same data sitting in there for a long time you index eight you quay at many times and you can build up statistics and understand information about the data like how popular different keys are and then you can estimate you know when you get a Nuke way you can estimate how much it will cost to do with different plants and pick a good plan the problem is with big data it's often unindexed you just got the data today it's like in some weird you know JSON or log file format there is no index so you don't have statistics on it and second the code running on it isn't these nice simple expressions all the time there's lots of weird user-defined function someone just gives you a blob of Java code or Python code and you can't easily know whether you know what behavior it will have how much data it will output for each record and so on so it's hard to actually get a good plan with cost based optimization especially in in more complicated ways so the solution is the let's park change the query plan adaptively this is actually something that you know has been done in in in in in research systems at least in the past and probably also in in systems that are out there it's something that we actually built even in the original version of shark which was sequel on spark and basically the idea is as you execute the quay look at the data that comes out from each step and then change the plan based on the statistics you see about it as you are computing it so I'll show a little bit of how that works so this is a simple spark program and this is the traditional scheduling that happens today basically we see you know a bunch of functions like mapreduce by key and so on and then we decide on a graph of tasks to run on the cluster so for instance we might say there are four map task because there were four blocks of data in the HDFS file and maybe we said okay the default number of a juicer is three and so we do the reducing the sword with three tasks but we don't really know what will happen you know this is a word count and sort so maybe we do the word count and actually there are super few words and they could have fit on just one reduced ask instead of the--of them but we don't know that until we actually look at the data or maybe there are actually many distinct words and sorting them is more expensive and we want to use more machines for sorting so in adaptive planning we change this to instead one the stages one by one and then look at the output statistics before deciding what to do next so when we do map we're going to create actually many different output partitions from the map as if there were lots of reduce tasks and we're going to write data at each of them so here I made five partitions and then we tell the scheduler how much data was written to each partition each hash code so you know maybe there's a lot of stuff with one key very little with another key and then the scheduler decides on a way to split these between the actual reduced s so in this case you might say you know there's actually not that much data so let's just have to reduce tasks and the top one takes two politicians the bottom one takes fee because that balances out the data and then we send that in after that you know we do the reduce function here it's kind of an easy one but imagine it was you know something more complex that could emit lots of Records and we do the same trick we write out file with law of different partitions and then we look at them and see the sizes of each one and here we might say okay the first three partitions are pretty empty but the next two are really big so let's round three three tasks and the first one will take three of the politicians and the next one's will take one each so basically we can balance the how the data is split into the next stage after seeing statistics about it instead of having to guess in advance and we can even change the number of tasks and stuff like that so pretty you know pretty kind of simple way of doing this but but it's actually quite useful when you do it and the final thing we can do with this is not just you know setting the number of tasks and like the ranges of keys that they're responsible for but even things like the algorithms use so for example the main thing like basically eighty percent of database optimization is about joins joins our when you have two data sets and you want to bring together data items with the same key from both of them and there are many ways to do it and the best way depends on the amount of data for each key so in with adaptive planning and spark this is what we could do we could take both datasets and again choose a say a larger number of partitions and create you know output from each one so counts how much data falls into each partition and then we can choose different ways of actually joining them so for example if if we think that this roughly balance data equal amounts of data for each key we can do a shuffle join which is actually the default in spark right now and in a shuffle join we just bring together stuff with you know different keys to different machines using reduce functions so there's one machine that handles everything that's blue in partition one there's one machine that handles everything that's red and there's one machine that handles everything that's orange and all the data has to move across the network but you know as long as any one machine isn't overloaded this is actually a pretty good way because then all the joining happens locally are on just the one machine when we bring together the stuff with the same key so this is one way in other cases though other algorithms might be better so for example if we hung these desk and we see wow the top data set produce very few records but the bottom one produce tons of them then we can use something called the broadcast join in that one we don't shuffle the ones at the bottom at all we keep them on the same machines and instead we take all the values at the top and we have placate then we broadcast them to every machine that had partitions of the bottom one and if this one is small this is still good because it avoids sending all of this stuff over the network you just send a smaller thing over the network many times and this is you know this is one of the things many distributed sequel systems do this if one of the data sets is small but you have to know in advance that this thing whatever map function whatever was earning him there will actually produce something small and finally with the way we're doing adaptive quay planning and spark you can really have any graph of dependencies between the tasks in this stage and the next stage and you can actually do some pretty interesting hybrid strategies as well so for example one thing that could happen is in the blue keys here maybe the data for each blue key is pretty small up here but in the bottom ones there is a lot of data for each key like for example the key is something like null or it's just a very popular key and the data so then we could do broadcast for the blue keys where we keep this stuff on the same machine and we broadcast the ones up here to those machines and then we could do shuffle for the other keys where we bring them both together and do the joint locally and avoid both casting them so this is a pretty you know pretty kind of interesting hybrid strategy you can do and it the way we were building this in spark you can have any of these communication patterns between you know the next stage and the previous one so you can actually do these things so this is this whole adaptive planning thing is something that we hope to put in the spark one that six it's in progress you can see it online at you know this gia 9850 so one of the last four digit jurors that we have you can check it out on it so that's that's pretty much what I wanted to talk about so you know spark as a whole it's going a lot of course in the few years but it still remains a highly active open source project and it's still as far as we can tell the most active open source run in this space and the reason for that is that there's a lot of stuff to be done both in the libraries on top of it and in the co engine because we have a very small core engine and this high level API that ties it to all the stuff above we can make changes quickly that affect you know every library and everything that's available on top and some of the changes that I've talked about that already went in our examples of that where we rewrote kind of a whole layer of spark and it's transparent to all users and it gives you these nice benefits and because there's so much interesting new hardware coming out these days and and you know basically new environments you can hunt with there are also a lot of interesting optimizations to do at all these levels so we're excited to continue working on them and to actually use you know the types of of machines that will be available in the future finally if you want to learn more about sparkle like learn how to use that if you're new to it I just want to send one you know kind of shout out to this community site so we created this community site on it we have quite a few different resources on it including two massive online courses that we put together on spark and links the lots of resources available off the site and you know you can check it out to just see you know all these resources about spark in one place so that's it thanks you