scale.bythebay.io: Matei Zaharia, Composable Parallel Processing in Apache Spark and Weld
Recording: scale.bythebay.io: Matei Zaharia, Composable Parallel Processing in Apache Spark and Weld
you Oh [Applause] let's see yeah thanks a lot so I'm going to talk today about making it easier to do parallel programming this is something we've been doing in the spark project for a while since we started the project and also something that that I believe we can continue to do even better not just for distributed coasters but for emerging parallel hardware and many other settings as well and the key idea I'm going to focus on this composition which is something that you know we've learned and we've been iterating on since we started the project so I'll talk about what's happening in spark also about the research that I'm doing at Stanford on new systems that do this so if you step back and think how why or how are people productive writing software what is you know the trick to make you productive writing software there's basically only one trick and that one is composition that's really essentially the only way we have any chance to tame the complexity of writing software is that we break the problem into little pieces and we compose those pieces to to solve you know the specific tasks that we have and that's essentially the only thing that we've had for you know whatever 50 60 years that we've been doing software development and it's kind of well-known but you know maybe people don't think about it all the time this is one example like why do people really like languages like Python and well you know it's not just the language but just you can kind of import anything and anything kind of works together you know functional programming person may say because they don't look at the types they just try to make it work but it is actually pretty useful or another example is it's programming via stack overflow so actually in Visual Studio there's a plugin that lets you as you type in what you want to do it just finds the relevant stack overflow snippets and you can just cut and paste them to put them in there and this is one of the best you know most productive IDs that that we've developed so that's composition everyone like said everyone uses it every day and it does make you more productive but unfortunately when you go to two parallel computing in particular and I also talked about other areas where this is happening at least the early AP that people came up with didn't really try to support composition as a first class concept and this created a lot of pain and of course a lot of data engineering tasks that you have to do just to build up your applications so in particular if you remember kind of the days of MapReduce and all the sort of dozens of parallel processing engines that people came up with you had two issues first of all they were just separate specialized systems for each task so you'd have to you know use one engine to do sequel queries and other one to do machine learning another one to do just basic data transformation and second between these engines you know a real application will need to use several of them and between them you had pretty slow data sharing they all said oh I just write your result to a file and pass it into the next one so if you look at Apache spark and what I think the single most interesting thing that it did and the reason that you know it grew into a large open source project with a large ecosystem of libraries is because it's supported composition so spark provided a unified engine and API and there were kind of two pieces there was first of all the engine that could actually run all the types of computation that you're doing but if you look at that it's not that different from MapReduce it's basically MapReduce plus in-memory data change so it's not like rocket science to do that and then there were composable api so these functional programming api's were you can hook together these functions and and crucially you also get efficient execution underneath so in particular the api in in the original spark was this thing called rdd's which is just as distributed collections of objects with functional operators and you know you could you could do things like map and filter and so on but the crucial thing underneath that actually made it possible to write stuff as lots of libraries is efficient composition so when you do these operations like when I do the map and filter here SPARC actually pipelines the execution so that you know each machine is reading one element doing the doing the filter right away it's not saving stuff in between as a file and this happens even if these lines instead of being right next to each other in the program are coming from two totally different libraries that I imported that were written by different people who've never met so for example this thing might be the map might be like the spark input source from some kind of storage system and then the filter is my own code and then within these we also had a model for sharing data which was just to have objects you know just give me a Java object and give me a pointer to it and I'll work with it so how well did this work so I think it in general users really like this kind of unification makes it much easier and simpler to build applications but the way we did it like the actual composition interface was somewhat limited so this original API cost some challenges and these are the ones were addressing so basically in this talk I want to first of all convince you that thinking about basically these software frameworks in terms of composition especially efficient composition is a good way to do it and then I want to show you some of the new ideas in this space that go beyond the RDD API and even beyond the other API is in spark today so first of all like do people actually use composition yes people built a whole bunch of libraries and most applications use multiple of these libraries these are just the ones that I included in spark but then there are kind of many other open source packages out there and then as you compose them you get these nice benefits in terms of efficiency so in terms of the you know in if you compare using these libraries and spark versus just running separate frameworks you don't have to write the stuff to a file in between you can just on a pipeline sequence of functions in memory and so you get you got high performance this way or like higher than you would certainly in any in other ways but there were also some challenges so the the main challenge is that the the kind of function API I talked about so far it looks high-level but it hides a lot of the semantics of the computation from the engine and so the engine can't always do Optima across your code that might make sense to do so there's two reasons that it's it's it's a little bit difficult to do that the first is that the functions are just arbitrary blocks of Java code so I mean technically we could imagine analyzing the code and figuring out stuff about them but that's both very difficult and very brittle you don't want your application you know regressing 100x in performance because you changed a couple of lines of codes that work and also the data is stored in these Java objects and that's kind of the model we have that's part of the interface so we can't do too much to make that more efficient and the result is that users can mix the API in suboptimal ways so for example this is one really common thing that we've you know that we've seen which is user is trying to do a word count but using the the group by key operation so this seems like maybe a nice way to do word count you you know you map each of these things into key value pairs and then you call group by key and then for each group you compute at some that's seems pretty reasonable and you know looks very concise and and nice but the problem is the group by key operator if you look at that operator its signature is defined as returning a sequence of objects here a sequence of ins and the scala sequence class is a collection that's basically materialized in memory and so just because of the signature of this operator we have to actually construct a collection in memory and return that and that's really an efficient because for this particular task of word count it's actually much if we knew we were just going to do a sum it would be better to do the aggregation on each node force maybe and then group them together or like to not materialize this in memory but start adding them up as we read them so that's that's something that we lose out on and you know and and and the engine can't do much about it because we said our interface is a tell us a Java class really turn and give us the Java function and we'll run that for you now this example might look kind of you know kind of silly like ok who's going to write code like and I do agree that yeah maybe an individual developer won't write this or will at least you know debug it and figure out how to fix it but remember what I said at the beginning which is composition so imagine that these things came from two different libraries there's one library that maybe does the first two lines it gives you you know the groups of work and someone spent time writing that and you want to use it and then there's a second library you wanna call to do this up it would be awesome to write your program as just I'm calling these two libraries but now you don't have this optimization so this is why this kind of optimization is important so what have you been doing about that we've actually been doing a bunch of stuff in SPARC over the past I'd say three or four years to to attack this problem through these higher-level api's called structured api's so I'll talk a little bit about those and then I'll talk about going given beyond this to support more interesting types of composition in different domains through this research project at Stanford called world so what are the structured api's well a new set of API is for SPARC that operate on what we're calling structured data and what we mean by structured data that is that it has a limited table like data model where you've got hoes with fields and so on it's very similar to you know the data model in a database or to the hive data model which has nested objects or to something like ever as well so the point of this is the data model instead of being Java objects like you know class whatever my dot company dot person they're actually these objects with fields that the engine understands and data types that the engine understands and then we have several different api's that all map down to the same kind of execution engine that spark sequel which you can use it to write sequel directly but there's also data frames and data sets which are more for programmers and they look more like the RDD api and then these things support much richer optimizations across your application similar to the ones you do in in a relational database but they try to be the high level of program ability that you had with with spark so how do these things execute well whichever of the api's you are using we build up what's called a logical plan which is a you know just a concept in query optimization it's a tree of operators and then we optimize it we look at what data is available to build a physical plan and then we actually generate code that uses rdd's underneath to actually execute this and then the final piece of the structured API is the data source API which allows us to plug in data sources and also actually to push part of the Quai into a data source as well which is a cool feature we can do here because we actually understand what the users computation is trying to do so I'll just give you a small example of data frames imagine people working with sparker have maybe seen that already so data frames are these distributed collections that hold the hose with a known schema and the structured data model and they offer relational operations on them so the kind of operations you do in a database through a DSL so that's a part that's different from rdd's so you can create a data frame using sequel if you want you can create it in other ways too so and here you know we're just connecting to a table like say a hive table that we have and then you can do operations on it like filtering I'm showing the syntax in Python because it looks a lot like the Python data frame library called pandas and the thing if you look in here this this stuff in the middle user state equals California this is actually this DSL we have this domain-specific language that's building up an expression and this is just an expression that means take the column called state and compared to California and return like you know a 2 or false if it's equal so this is just the abstract syntax tree for the expression we don't actually evaluate it we just have a way of building it up if you compare this with rdd's and rdds we had to give you a Java function or a Python function that you execute that that gives you the two or false so we didn't understand what the function does we just all we could do with it is run it and see what it gives whereas here we actually the inner code in this kind of mini domain-specific language that we understand and then you can you know you can do other things like count these ones after filtering them you can also do grouping and an aggregation one note here is that if you look at this group by an aggregation you know the the basically they actually reference the field names used in there so so the engine actually knows which piece of each record you're looking at as opposed to just having to bring all the records together and and and on a function to figure out what you're pulling out from them and then if you know if you can't do things using the DSL you can always fall back and write your own functions over them it's just it won't be as efficient or like this will be a boundary where we can't do as much optimization so that's what it looks like now what do these API is enable so because we can understand the data model and because we can understand the expressions we can do three pretty nice things underneath them so first of all we can have a compact binary or presentation of the data we actually store data and a columnar format when you're caching at long term we store it in a row format for processing but it's never it's never like a giant graph of Java objects with pointers between them so it's actually just more efficient this store we can do optimization across operators so for example we can push you know a filter before a join because it's only filtering one thing and so on and we can do runtime code generation so we can actually generate specialized code for the data types that you're using in your particular program no matter how you put together that program and of course as you use this it has a lot of the other nice properties that already decide you know you can embed it in like a full programming language you can use classes and functions and if statements and stuff like that to build up your computation so it's it's nicer than just writing sequel for example as well so how does it perform so because the the structured engine can specialize for specific schema and your specific computation it often can actually perform better than rdd's of course depends exactly what you're doing but if you see and hear the data frame API at least in this kind of aggregation benchmark it did a little bit better than than rdd's in both Scala and Python and it didn't matter what programming language you're using because underneath were compiling everything to the same kind of optimized code so that's another nice thing it usually in spark before this you had to worry about like oh if I use Python things will be a little slower because I'm just calling into Python all the time so this is this is just briefly what this gives you and we we also offer kind of other API is on top of this so I talked about sequel another API we offer is called data set which adds static typing so kind of like so basically you have your data frames as before but you say okay the schema of the data and this will correspond to this job our Scala type so as an example let's say we've got two classes in our program you can use case classes in Scala to represent your data let's say we have user and message we can read a data frame using just the standard spark api's for that and and then we can tell the system hey we want to view this as a bunch of message objects and actually add analysis like when you're running this code it will look and see whether the schema matches that so it will actually tell you that hey these things don't look like message they don't have the fields user and text and then we can do operations on them where we view these things as Java objects so so you know things like filtering and mapping where we just view these and actually spark preserves the datatypes to this so for example when we map you know m to M that user we go from a data set message to a data set of user and if we want we can also go back and use the full the full DSL available on data frames and and do things like go by and just specify this string in here so this is a nice way to combine kind of the benefits of the statically-typed you know Java object type API and our duties and of this one in particular if you're writing a big piece of code you can document all your functions as returning data set of a specific type so you have you know you understand the typing as you work with other people on your team everyone agrees on what what's going on in the data and you can still use these these more optimized kind of operations underneath okay so that's just a brief overview of why these work hopefully it gives you a sense of like what are they enable that's that's more powerful and since we started this work I think this started around 2014 we've actually built a whole bunch of other API zhan on top of these structured data frames and assets so for example data sources for spark machine learning pipelines and and graph processing and structured streaming so I'll just show you a couple of examples of what you can do and why it's powerful so data sources this is the standard way to connect spark to a storage system now and basically the the part that's interesting about this API is that it allows quai Federation into the sources so as you're writing your computation for example and we had that filter expression before spark can figure out which filter is applied to this specific data source and it tells the data source here are some filters for you and if your data source understand these it can actually push them into the storage system and avoid reading all the data into spark this is something that would have been very hard to do with the original system so just as a simple example let's say we've got a JSON file you can view it as a bunch of records using using spark and have fields like user ID and so on let's say we also have a my sequel database and we want to select some stuff from our JDBC and let's say that we're actually trying to combine these two data sources so we're trying to combine appropriately enough for Twitter it's about tweets so we're trying to get the tweet text and the user age but the user ages in a database and the tweet text is in some kind of JSON file because that's what Twitter gives us and you know and we want this filter and use their language equals English so actually spark can look at this query on this side okay from the database we only need ID and age and we want to only filter on language equals English and put that it and again if you go back to you know what I was talking about with combining libraries don't think of it as like one developer knew they have to do all this in advance but think of it as someone hold a library to like say load the users for you someone else or the library to analyze tweet and you're trying to hook these together and still get good performance without having to dig into each of them and and plan out your your whole execution so that's what it does and then the the kind of the second example and I want to talk about why this is powerful is structure is dreaming which is the high level streaming API built on data frames and data sets that that came out basically last summer 2016 so this is just this is kind of a higher level API that tries to keep all these properties the query optimization and so on it supports things like event time processing windowing stateful operations and one of the goals is to support these end-to-end what we're calling continuous applications where you're not just you know receiving stuff in a stream doing a map function and writing it somewhere but you're trying to write a complete application that might involve interactive queries batch jobs and so on so it's designed to make it really easy to interact with the other parts of SPARC to do that now the really cool thing about this is actually the API so we added this whole streaming engine but we didn't really have to add a new API instead we the way we do it is that we incremental eyes an existing way you have using the structured api's so as a small example imagine we have this batch job here this this thing is just a batch job it's trying to read some JSON data from HDFS and then group it together by by the user ID and also the our field so this is like the event time when when something happened and figure out the average latency and let's say we're trying to write that to a file in an ST so that's kind of nice we can understand what the job does and and save it if we want to make this into a streaming job that updates the result incrementally as new data arrives we just have to change the input and output method so from read to heat stream and same thing for right so all we have to do is say ok well we're now going to read a stream just keep looking for new files in this in this location and then we do the same grouping and then we're going to write to parquet and and we're going to actually update the counts in this bark a table over time so inside you you run the same operations as before and because SPARC understands the expressions that you're using in there it knows how to actually incremental eyes the computation for all of them so what it'll do is it will take this batch query plan that had a bunch of you know a bunch of operators and just transform it into an incremental plan that includes keeping track of State and updating this thing I guess my sequel or SSD updating that transactionally so that you you see the right values appear so it's a pretty powerful API to do streaming it's really easy to use if there's anyone in your organization who knows sequel or data frames they can use it it's really easy to test because you just switch it to the batch mode and and you can just run your code and get a result and it's only really possible because we can look inside these operations and understand them after you've put them together in your program just as examples of this this is a pretty new API but we've actually seen very fast kind of growth in usage so at data breaks which is the startup company I co-founded we we own a cloud service so we can actually measure and see how many people are using the API and there are already hundreds of customer applications running and production so these are things running 24/7 the largest of which process tens of trillions of Records per month some example applications that people have given talks about are monitoring live video streaming delivery on most of Viacom's web properties anomaly detection a type s which is actually the world's largest operator of Wi-Fi hotspots and also real time analytics for league of legends from hired games and the cool thing across all these use cases actually the thing I'm most surprised by is how few issues we see in terms of support tickets and so on because people if they already know how to use sequel or data frames they can write something kick it off and it just runs and and they don't need to learn a new API and get confused by it as its as it's going along and we've also spent quite a bit of time optimizing the performance here so because this engine builds on spark sequel and on tungsten where we spent tons of time you know making it competitive with with kind of analytical data warehouse it's actually pretty fast for streaming as well so for example we ran this Yahoo streaming benchmark that a bunch of systems optimized for and just out of the box we got about 4 times higher throughput than the next fastest result that had been published for this which was far more patchy flink and we didn't have to do anything special to optimize the code here it's just the code for doing aggregation and selection and so on in in spark sequel was already optimized for that so the really cool thing about this for streaming is of course if you're streaming you're running something 24/7 so if you can hunt the same thing with four times fewer resources that can save you lots of money so that's pretty good and there's actually also an operational benefit if you can run the same thing on four times fewer nodes it's four times fewer things that fail four times you're things to upgrade and so on so we think actually basically poor node performances is probably even more important and streaming than it was in batch processing because batch processing you're often bound by just reading stuff from from some kind of slow storage so that's that's that's one other thing that if you want to look at what's happening with the structured api's this is where there's a lot of activity today okay so in in the final bit of the talk I want to go a little bit beyond this and say how I'm I'm taking this idea a little bit further in research at Stanford so this is not necessarily stuff that's that's in spark where that's kind of production-ready but it is interesting to try to take this further and partly from seeing what people were able to do with these api's in spark i've been very curious to see how far we can push them so we at Stanford we've started this project called weld which is trying to design the same kind of composition based efficient programming interface but for modern parallel hardware we're really targeting things like multicores GPUs and potentially programmable hardware as well so the motivation is that with continued changes in hardware actually even your local machine is a distributed system and going to memory is very slow so memory is sort of the new HDFS if you're just putting stuff in there it can be much slower than your computation and this means that the traditional interfaces we've used for composing libraries just on a single machine are increasingly inefficient actually we see this with spark workloads as well often we get things that are CPU and memory bound on each node it's it's not the network that's a bottleneck so what do I mean by traditional interfaces well so the traditional way you compose software on a single machine is basically through function calls this is something that we've had since pretty much since people started writing you know kind of structured programming languages and the the contract of these is is pretty simple is basically okay you you put some values in registers maybe and you jump to a piece of code and usually to pass in a large amount of data you have to put it in memory and and pass a pointer to it into the function so basically write your stuff to a memory buffer called one of these functions and not a ton but if you look at a lot of computations today they don't do a lot of work per byte of memory or more accurately what I should say is memory has gotten much slower relative to CPUs than it used to be when these interfaces were designed so as a simple example let's say I'm just doing some simple data transformation I'm just going to show this in Python but actually even in Python when you call these functions each of them is actually hidden in C so it's it's not you know the problem is really not the Python so let's say I've got one function that parses CSV files so I've got my data in memory and maybe I parser to get a table and then I've got another function to drop the NA values and I've got another function to compute the mean this is going through essentially the whole data reading and writing about five times whereas in fact you know much like the pipelining in SPARC if I just knew I would have to do all this in advance I could just go scan through at once and compute the mean in a running fashion and not have to do all this stuff and again when people first designed these programming languages it took you know took one or two cycles to access memory so basically if you if you hold a function and it had to go to memory it wasn't a big deal because you know you're gonna do some computation on each byte anyway so it's okay if you take a cycle together but now it can take basically the equivalent of hundreds of cycles and so the memory kind of transfer costs can dominate and if we look at workloads in today's frameworks we see that often for data intensive workloads this creates very large overheads in even in systems where each operator and each function is optimized by hand like dump I well you know some scientific program we sat down and I try to make each each operator really efficient so how can we fix this so basically what in world we're trying to build the equivalent of a long time for data parallel computation similar in some ways to JVM or to the SPARC sequel on time I talked about so we're going to have this this one time where people submit snippets of computation and then we actually optimize across all of it and produce efficient code for the whole program not just for these individual ones and there's a few important pieces to this so first there is the API that that libraries used to actually submit code and then there's this this common IARC this is a common intermediate representation for data parallel code this is kind of the equivalent of the logical plans I showed before but we're actually trying to capture more types of computation here and then we have an optimizer and we have these hardware backends so let me just start by talking about the API this is actually pretty simple conceptually it just uses lazy evaluation similar to spark to collect bits of work to be done so as you're running different functions in your application instead of executing eagerly they'll each just push a bit of work into the runtime they'll say hey I wonder on this computation and you know here's here's kind of a handle to the result and then when at the end you can actually force it to execute and then it will look across all of these and do something it's actually interesting to compare this with api's like OpenCL and CUDA today so people use these to get high-performance people right key parts of the code in OpenGL but all of them execute eagerly so there's no way if you have several different libraries using OpenGL in your application the system doesn't do any kind of optimization across them so you can think of this a little bit like a lazy evaluated version of that okay and then the second component is the ir so like what is kind of the language in which we submit code and this is where you know we tried to go beyond the kind of things you get in spark or in a database with a more general language that can capture more things so we have this really small but but powerful ir inspired by something called monad comprehensions and it just has two constructs has got parallel loops which let you iterate over a data set and it's got something called builder is which lets you produce the results so again builders are kind of abstract data types so an example of builders is you can have a builder that adds objects to a list or you can have a builder that computer some but we can implement these differently based on the target hardware and we abstract that away from the program or how exactly will computer some say on a GPU so just as a few examples of implementing stuff using builders I'll show how to implement some functional operators so let's say we want to implement map we've got a bunch of data and a function f we can basically we can create this thing called vac builder which is which is gonna append stuff to a list and then for each item in the data we can merge F of that item into the into the final list and this is this this loop happens in parallel and actually the builders are all associative so it preserves the order of these things and then the the last operation you can have on builders is called a result which takes that finalizes it and gives you back a value so that's what it looks like if we want to do hee instead the code actually looks super similar the main differences our builder instead of being this thing that makes the list it's something that merges values with with an associative function and now for each item we merge it into there and again we call the result and we'll get that back now the cool thing about this representation is it's also easy to represent fused operations and just basically ways you're going to optimize this execution so for example let's say we have a program that's computing two things over the same list you want to compute a map to build the squares and you also want to compute a reduce over the the same data item if you were just going to write this by hand you would probably just write a single loop that does both of them and in world if we see the bits of code that came from what I showed earlier we can automatically fuse them to do something like that so it's easy to represent that and beyond this we can actually do a bunch of other optimizations like loop tiling and vectorization that also greatly improve memory efficiency and require this kind of n/2 view so how does this actually perform we tried really kind of rough prototypes integrating this into different systems and we see that just by adding weld underneath the existing API so without changing the user facing API we can get somewhere like a 5 to 10 X speed up over the naive way of running these things for tensorflow I don't show it here but the hand optimized code is also similar to the tensorflow excel a compiler so so are pretty small and general language can also match that and for numpy and spark we got about a 5x speed-up and then if you have a workload that combines many systems there's no way that today you'll you'll get great performance even if they each implement their own compiler but with world you can actually do that so this is showing weld without cost library optimization gives you about a factor of 9 speed up with cross library it's a factor of 30 and then welders also the language is explicitly parallel and you can run it anything on multi-core so if you run it on multi-core you can get a for the speed up to so overall we think there's a lot of space to greatly improve the performance of current tools and to keep programmers productive so we're we're working to build this out and and also build a bunch of libraries on top of it if you want to find out about it it is implemented in rust as alexey said I can talk offline about you know what's good about that but it's actually kind of kind of fun to work with and it uses LLVM for code generation and we've open sourced world as well as a port of pandas thrown on top of it so you can check it out online and then the final thing I want to just briefly mention is we're also using this approach in or the research project so for example similar to the data source idea I showed for SPARC we have a system called sparser which generates parsers and ioq paths for for your specific way this actually goes all the way down to generating sim D instructions on the CPU and and trying to vectorize it to make it very fast and we see pretty large speed ups there as well so we think this idea of assuming that developers will just want to compose stuff and then trying to design interfaces that let's us up optimize under those has a lot of legs and and we're trying to do a bunch of stuff in my group to take it further ok so that's pretty much what I wanted to talk about basically in short if we want to be as efficient working with modern hardware as you were working with a single CPU we have to change the way we compose software and it's the only trick we have to make it you know not rocket science to do that and Apache spark and weld are just two examples of projects that are going to watch this go Thanks [Applause] do we have questions or do we I don't know how this works here yeah do we have time for questions Alexi okay yeah sure go ahead let's kill immunization or caching of the video are you thinking about that or do you have any pointers yeah that's a great question yeah actually we do we try to do a little bit of that in Weld but we're not doing that we're not doing it in a super principled way but you see workloads where you compute the same thing over and over and we could analyze it and figure out okay let's save this thing once yeah it does it definitely can help and it happens by itself yeah yeah at the back which is - I miss y'all and eventually yeah that's definitely one of the goals so it has kind of a couple of goals one one is to decide if if we can even get good performance from from composing libraries this way and we think we've shown you can get a bunch but then I'm also really excited about the way we abstract over a different hardware in particular the concept of builders basically when we looked at how do people have to change their code to go like from a GPU to a multi-core to like a Xeon Phi or something the hard part is usually the synchronization and that's all encapsulated in the builders here so like we think that actually it's not too hard to make this discord portable across T's and we're working on that as well and the kind of the cool thing with that is we think it's like super important for the industry to have a standard IR to do this stuff like know what you can truly innovate in Hardware today without convincing people to switch software to use it and this thing would allow you to innovate and hardware much faster yeah okay thanks [Applause]