1st Spark: Matei Zaharia introduces Spark and RDDs at Scala for Startups hosted by Alexy Khrabrov
particular data flow. So basically you set up a graph of operators and then you load your data from a stable storage system like Hadoop file system and you pass pass it through this graph and it comes out at the other end. Okay, so this this isn't really an accident. Um the reason they use data flow is because it lets the runtime do a lot of things automatically for you so that you as a programmer don't have to worry about them. So the runtime can decide where to schedule the tasks. It can decide how to recover from failures. It can decide what to do if a node is running slowly. All that kind of stuff and you don't have to do it
Okay. But the problem with data flow is that it essentially only gives you a computing um abstraction these different operators and it doesn't really give you an abstraction for um for memory for keeping data around between the the uses of different operators. So it's actually inefficient if your application needs to reuse data sets. So, one example of this is iterative applications, things like page rank. Um, if your data is something that could fit into memory across your whole cluster, then you could go much faster by keeping it in memory rather than dumping all your state to HDFS and loading it back up on each quay. And actually, we found a lot of applications where when you run it with Hadoop, more than 90% of the time is spent just doing IO and serialization because it has to pass through this file system. Um and the other one is of course interactive data mining to get interactive speeds you it's very hard to do it with on disk data but if you can u have the tools to keep the data in memory you can go much faster so with with the current tools like Hadoop you can't really do these these things without just passing through a storage system every time so what we provide in spark is this programming interface or abstraction called resilient distributed data sets or RDDs and you can think of these as distribut ed collections kind of like your lists and arrays in Scala except that they're spread out across multiple machines and they're fall tolerant and with Spark you can choose which RDDs to keep in memory they'll stay in memory as Java objects so there'll be no serialization cost or any of that stuff across the queries you had on them um and at the same time you get the nice properties of map reduce it can recover from failures it gives you automatic scheduling data locality and high scale and we also just showed that with RD s you can actually write a lot of parallel algorithms efficiently. So you can do a lot of applications
So in this talk I'm going to go a bit through the programming model, show you some example code um and then talk very little about the implementation. Um the the other thing I want to do is also show you a demo of this happening interactively. Um and then we we actually have uh Dillip and Tim who will talk about some applications they've been doing uh large scale you know data analysis and machine learning. So let's start with this one. So in in the programming model there are sort of two things you need to know about. First are the RDDs is distributed data set I already talked about. So these are immutable um distributed collections and um the way you build these up is you can start an RDD from say a file in Hadoop or you can apply these operations called transformations. And transformations are things like map um filter um group by join
we actually have a wide range of transformations not just map and reduce. Um the other thing you can do is you can tell the system to cache some of the RDDs in memory because you want to reuse them. Um all the RDDs that you make are built lazily and then the way they actually get used to to produce an answer is through another set of operations called actions. So actions are things that return a value to the program who called them or that actually you know save data to an external system. So there are things like count returns a number how many elements are in the data set um and so on. So I'll just show an example of of how this works in in one application. So in this example um this is stuff you can type interactively from spark and we'll be analyzing some console logs. So say you're running you know big uh successful web application and uh something is going wrong and then you have these hundreds of gigabytes of logs that are being written say to the Hadoop file system
in the application. We're going to load the error messages from the logs into memory. Um, and hopefully, you know, this is actually a small fraction of the data. And then we're going to run uh queries on them. Uh, so this is a picture of our cluster. We're going to have one node, the master or the driver program is what we call it. That's where you're running your your code, your shell. And then we're going to have some worker nodes
And the first thing you type in is is this following. Um, so hopefully you can see that. So we define lines as a text file um in in HDFS and lines is going to be a distributed collection of strings one per line in the file. Okay. Um so this is the base RDD that then we're going to transform. Next thing we're going to do is call filter and we're going to pass it the stuff in red as a Scala closure. So you can have any code you want in there any Scala code, any Java code. Uh here we're just filtering the strings that start with error
So we're pulling out the error messages. Um so this this gives us the transformed RDD back. Um we might do more transformations after. So here we're doing a map and we're taking um each um each error line and we're it's a tab separated line and we're pulling out only field number two which might be the actual message in in the log. And then when you're done building the data set that you want to use, you can call cache. So that will tell Spark when you've built these message objects, keep them around because I'm going to be asking many queries about them. So so far nothing's happened on the cluster yet because this is all lazy. Um you can then you know start start running these operations that are actions
So here we're going to count how many of the of the messages contain fu. Um and this is this is an action. So so now Spark actually starts to run stuff on the cluster. Um, so first it figures out where the blocks of the file are and it sends tasks to the nodes to process those blocks exactly the same as map pages. So trying to maximize data locality. Okay. Um hopefully I'm not in the way of people here. Yeah
Um okay. And then um uh the workers process these blocks and send back results. Um and the workers can also cache any partitions of of cached RDDs that they've built along the way. So now you have just those error messages into memory and so you get back a result. Now next time you had a question on this data set, maybe foo wasn't the problem, maybe you want to search for a bar. Next time you run it, Spark will know that this data is in memory and it will schedule the tasks according to, you know, to memory placement. Um, and you'll get back an answer much faster. And of course, you don't have to just do filters and counts
You can do all those group pies and joins and other kinds of operations that I said we have. Um so just to give you a sense of of what a difference this makes um one of the tests we do is um just a full text scan of a of Wikipedia just searching for something in it u you know using string that contains um and when we run it on 20 extra large EC2 machines it takes about 20 seconds with on disk data or with Hadoop and it takes less than a second with in-memory data with Spark and of course this is because uh memory is a much higher bandwidth storage system than disk And because you're avoiding all the overhead of serialization and stuff like that that has to pass happen as as you pass through HDFS. Um and just for fun we've also scaled this up. So the biggest we we've run this kind of demo on was 100 of the high memory you know extra large machines on EC2 the very largest type. And we we loaded one terabyte of data and there we could still do a full text search in 5 to 7 seconds. Okay. So that's that's kind of what you can do with it. Um the other cool thing about the model is the way that we get fault tolerance
So Spark knows how each of these data sets was built through the different operations, the map and filter and so on. And so if it loses a partition of one of the data sets, it can actually rebuild just that partition. So for example here, if we lose this cache over here or say this node, Spark will find another node that has a copy of block three and apply the map and filter on it and rebuild just that block. So you don't have to go back uh to the beginning of your program to to get fall tolerance. Um so this is this is we call this lineage. So we track the lineage of each data set how it was built through the transformations and then we can have an efficient plan for recmp computing each partition and inside spark we just represent each data set as a graph of of these transformations. Okay. Um let me give you a second example
This is more of a iterative application from machine learning. Um so this is logistic regression. This is something you might use for example to find spam on your social network to classify uh two sets of points. Um so the the story here is you have these points um in a high dimensional space your features and you want to find a a line or hyper plane that best separates them and you don't know what the line is. So the way the algorithm works is it starts um with a random line initially um and then it applies this this gradient descent uh to try to to improve it to separate the points better. Okay. So at each step um it computes this gradient function sort of the derivative of the function you're trying to optimize which is how well the line separates the data sets and it's this is essentially a sum over all the points and you can see it as the points pushing and pulling the line in some direction. And then you you can move the line a little bit by updating the parameter based on the result of that
Um and you keep doing this for multiple iterations and eventually you converge to a pretty good separating line. So um this is the code for doing this in Spark and again this is actual Scola code you could type in. Um and there's sort of two pieces. Um the first piece at the top we're building a data set of the points and because we reuse the same points in every parallel sum of the gradient we're going to keep those in memory. So we call cache. Um and then in the second point we have these iterations and on each iteration we do a map and a reduce to add up this function that is the gradient function. And you can see actually one cool thing here. So w is is the current parameter
we have the current line separating the two and we actually use the the value of W inside this function that we're running on the cluster. So the Scala compiler and the Spark runtime will make sure to ship that variable for you along with the the closure to all the nodes in the cluster. And of course if if you changed if data here was just an array or a list in your local program, the same code would work in the same functional style locally because it's the same way you program with Scola collections. Um and again when we compare this with Hadoop um we we can get quite a bit of a speed up. Um so this is another experiment on on 20 EC2 nodes. Um and here if if we ran this logistic regression with Hadoop basically Hadoop takes the same cost on each each iteration because it's doing the same thing. It's loading the the points from HDFS. It's parsing them into Java objects and then it's actually doing the math
Um, when we ran this with Spark, the first iteration took a little longer, but the next ones were about 20 times faster. And this is again because in this job, the math you do for for each point is so little in terms of compute cycles that the time is dominated by IIO and by all the other stuff. And so if you if you want to use this at the full speed of your cluster, you have to keep the data in memory. Okay. So so these are kind of, you know, these are kind of um example applications I can put up on one slide. Um, Spark is also being used for a bunch of real applications um, both at Berkeley and uh, at at a growing number of startups actually. So I I' I've heard you know from a couple of people that that have started to use it. Um, so at Berkeley uh, one of the ones you'll hear about is Mobile Millennium Project
This is a big project uh to to predict traffic conditions in the city from GPS data essentially from people's cell phones. and they do a very good job of doing this and giving you predictions for streets that for example Google Maps doesn't have the level um of of instrumentation to get um so you'll hear about that from Tim um at Conviva. Conviva is a is a you know video distribution and analytics startup and they've been using um Spark for a while to do uh in-memory analytics and anomaly detection on Hive data and Dillip is going to talk about that. It's also a pretty cool use case. Um at Berkeley we have another project called Monarch that that did spam classification um using Twitter data. So basically a much fancier version of the example I showed and we also have people using this for collaborative filtering for time series analysis and other stuff like that. So I'm I normally I would show you know some slides about the applications here but I want you to hear from the people who actually run them. Um so the other cool thing that we we're doing with Spark is essentially building other programming models on top of it
So if you if you want something higher level than the Scola you know map filter that kind of stuff you can actually implement a lot of cluster programming models efficiently using RDDDs and using spark. Um so map reduce I think you already saw you you have the map and our parallel reducers called reduce by key. Um dryad um is something that we can support as well. So, Spark internally actually runs a a general uh DAG of tasks. It doesn't just run map reduce steps. Uh and so you can build these kind of more general graphs that come up in say SQL queries and and run them efficiently as one graph. Dryad is Microsoft's uh you know generalized map agents that lets you do this. Uh another cool thing we can do is per sole is a graph processing framework from Google that initially looks very different from Spark because it's based on passing messages between the nodes
But it turns out that we can implement it efficiently using Spark and also have a lot of its optimizations like partitioning optimizations. Um and uh we have so so this was actually implemented by an undergrad Ankor who's over there. Um you can ask him about it. This is open source and it's part of the Spark codebase. It's it's a pretty cool thing you can use. Um and then the other thing that we're working on now that isn't actually released yet is Hive on Spark which we're calling Shark and this will be a part of Hive. So it'll be backwards compatible with hive but it will let you do the queries interactively with in-memory data. Okay
So just one or two words about the implementation um before I go on and actually show you this thing in action. Um so Spark is built on top of uh MSOS which is a project that um we developed earlier in our lab um for managing clusters. So for sharing resources between clusters uh and it's also an Apache incubator project right now. Um so Messos the idea in messos was we saw a lot of interest in new programming models for clusters and people had these big Hadoop clusters where they were stuck with map produce and we wanted to build a resource manager that can efficiently and scalably run more of these frameworks together. Uh and that also makes it easier to build new frameworks. So uh with MSOS you can run Hadoop, you can run MPI, you can run say two different versions of Hadoop on the same cluster and you can also run Spark. Spark is actually the first new framework we built on top of it. Um even though so so Spark is built from scratch
It's it's not sort of a part of or a modified version of Hadoop or something like that. But even though that's the case, we can use Hadoop's input and output libraries. So you can use any storage system and input format serialization stuff like that that you get with Hadoop. Um, and we also didn't make any changes to Scala, which is also nice because we we don't want to be in that business. Okay, how do we do the the language integration? This is like one of the more interesting things, especially if you've used Scala before. Um, so this is, you know, it's it's kind of a cool thing that that a lot of this is done by the Scala compiler. So in Scola all these closures like the split and contains and stuff like that that I showed are just Java objects and the compiler makes them serializable if you only use serializable types inside them. So in theory we could just save them on the master ship them to the workers and all the variables they depend of and stuff like that will get shipped
Um now in practice we actually had to do a bit of work uh to work around some cases where the compiler references too many variables from your closure. Um but the the good news is we can do this at runtime without changing the Scola compiler and it's actually pretty stable. Um and we have some other cool tricks like for example uh we have something like counters called accumulators that just let you call plus equal on it in your task and send the value back to the master. Um and then finally for the interactive spark we modified the Scola interpreter and this this requires sort of the most hacking but at a high level there are two things. One of them is making sure that for each line of the interpreter when you save a closure on that line, we have references to all the objects that it depends on so we can ship them along. And the other one was to actually send the classes and load them over the network from the other nodes so that the code can just run over there. Okay, so I'm going to show u hopefully um a demo of this thing in action and then you can hear about some of the applications. Um okay, here we go
Yeah, let's hide this guy. Okay. great. Um, so let's see. Um, so basically for this demo, I set up an EC2 cluster with uh 20 20 machines extra large and I loaded a dump of Wikipedia on it. So it's a tab separated uh file and then I set up Spark on it. Um and this is so this is the spark interpreter and we're going to analyze this data interactively. Um so spark spark interpreter looks a lot like the scholar interpreter with some asyard at the top and it lets you do the exact same things you can do in scalar
Um but the the thing that you have extra is this variable sc or spark context which you use to actually access the distributed um features. I don't know if people can see that. Maybe I'll move this back um up a bit. Yeah, let's do that. Okay. And maybe I will grab a chair and sit down. Yeah. Sorry
Okay, there we go. So, what we're going to do here, we're going to analyze this dump of Wikipedia. And let's uh so I just had that in my history. Um so I have this text file on EC2, which is Wikipedia. And this is a data set of strings. Um so first thing we want to do with it is probably just see what the data looks like. This is a tab separated values file and we can we can actually just take a peek at the first string by calling first. Okay
So, so this is you know this is one line of text in the file and it's tab separated. Um and there are actually five fields on each line. So first field is an article ID. Then there's a title. This might have been the alphabetically first thing. Um then there's a date. There's an XML version of the article. And you don't see the fifth field, but there's also a plain text version at the end
So in this demo, we're going to only work with the titles and the plain text. Um, so sorry about So let's make a class uh to represent the articles. Okay. So we'll have this. Okay. And now let's let's transform the file in from a collection of strings into a collection of articles. Uh so what we're going to do here is we'll take each line uh we'll split it by tabs and then uh some of the lines don't actually have the plain text field because they're things like images. So we're going to take only the ones with five fields
And now we have these five tab separated fields inside an array because we call split. And we're going to to create articles from fields number one and four. Okay. So this this is now a data set of articles. Okay. And let's let's try so so with each data set it can either be cached or not cached. Let's first try the the not cached one with the undisk data and see the performance of that. So let's try to run a query over this
Um and we'll just count how many of them contain Berkeley. Okay. So now Spark is starting stuff on this cluster. U this preferred stuff means that the block of the file is on that node. So it's getting data locality and you know it's going to take a while to run this stuff. Um but it is running this the the code that we typed in. Um, and it's going to give us back account. Okay
So, it took 19 seconds and it says about 13,000 articles include Berkeley. Uh, which is, you know, not not that bad, I guess. Um, so, so, so that was good, but that's not exactly interactive. That's, you know, especially once you get to a bigger file, that's at the point where you go and take uh take a break and grab some coffee as Alexi was saying. So, let's try to also uh tell the system to cache the articles. Um, and then let's try the query again. So, right now the the articles aren't in the cache, so it's going to take about the same amount of time. Um, but you see when we're running it now, it's saying that it's added these entries to the