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 it through this graph and it comes out at the other end. Okay, so this isn't really an accident. 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 abstraction — these different operators — and it doesn't really give you an abstraction for memory, for keeping data around between the uses of different operators. So it's actually inefficient if your application needs to reuse data sets. One example of this is iterative applications, things like PageRank. 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 query.
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 I/O and serialization because it has to pass through this file system. And the other one is of course interactive data mining — to get interactive speeds, it's very hard to do it with on-disk data, but if you have the tools to keep the data in memory you can go much faster. So with the current tools like Hadoop you can't really do 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 distributed collections, kind of like your lists and arrays in Scala, except that they're spread out across multiple machines and they're fault-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. And at the same time you get the nice properties of MapReduce: it can recover from failures, it gives you automatic scheduling, data locality, and high scale. And we also just showed that with RDDs 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, and then talk very little about the implementation. The other thing I want to do is also show you a demo of this happening interactively. And then we actually have Dillip and Tim who will talk about some applications they've been doing — large-scale data analysis and machine learning.
So let's start with this one. In the programming model there are sort of two things you need to know about. First are the RDDs — the distributed data set I already talked about. So these are immutable, distributed collections. And 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, filter, group by, join. We actually have a wide range of transformations, not just map and reduce.
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. All the RDDs that you make are built lazily, and then the way they actually get used 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 save data to an external system. So there are things like count — returns a number, how many elements are in the data set — and so on.
So I'll just show an example of how this works in one application. In this example, this is stuff you can type interactively from Spark, and we'll be analyzing some console logs. So say you're running a big, successful web application and something is going wrong, and then you have these hundreds of gigabytes of logs that are being written to the Hadoop file system. In the application, we're going to load the error messages from the logs into memory. And hopefully, this is actually a small fraction of the data. And then we're going to run queries on them.
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 code, your shell. And then we're going to have some worker nodes. And the first thing you type in is this following. So we define lines as a text file in HDFS, and lines is going to be a distributed collection of strings, one per line in the file. 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. Here we're just filtering the strings that start with error. So we're pulling out the error messages.
So this gives us the transformed RDD back. We might do more transformations after. So here we're doing a map and we're taking each error line — it's a tab-separated line — and we're pulling out only field number two, which might be the actual message 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 far nothing's happened on the cluster yet because this is all lazy. You can then start running these operations that are actions. So here we're going to count how many of the messages contain foo. And this is an action. So now Spark actually starts to run stuff on the cluster. 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 MapReduce. So trying to maximize data locality.
And then the workers process these blocks and send back results. The workers can also cache any partitions of cached RDDs that they've built along the way. So now you have just those error messages in 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 bar — next time you run it, Spark will know that this data is in memory and it will schedule the tasks according to memory placement. 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-bys and joins and other kinds of operations that I said we have. So just to give you a sense of what a difference this makes — one of the tests we do is just a full text scan of Wikipedia, just searching for something in it using string contains. 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 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 happen as you pass through HDFS.
And just for fun we've also scaled this up. The biggest we've run this kind of demo on was 100 of the high-memory, extra-large machines on EC2 — the very largest type. And 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 kind of what you can do with it. 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 to the beginning of your program to get fault tolerance.
So 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 recomputing each partition. And inside Spark we just represent each data set as a graph of these transformations.
Okay. Let me give you a second example. This is more of an iterative application from machine learning. This is logistic regression. This is something you might use, for example, to find spam on your social network — to classify two sets of points. The story here is you have these points in a high-dimensional space, your features, and you want to find a line or hyperplane that best separates them, and you don't know what the line is. So the way the algorithm works is it starts with a random line initially, and then it applies this gradient descent to try to improve it, to separate the points better.
So at each step, 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 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 can move the line a little bit by updating the parameter based on the result of that. And you keep doing this for multiple iterations and eventually you converge to a pretty good separating line.
So this is the code for doing this in Spark, and again this is actual Scala code you could type in. There's sort of two pieces. 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. And then in the second part 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 the current parameter — the current line separating the two — and we actually use 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 closure to all the nodes in the cluster. And of course, 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 Scala collections.
And again, when we compare this with Hadoop, we can get quite a bit of a speedup. This is another experiment on 20 EC2 nodes. And here, if we ran this logistic regression with Hadoop — basically Hadoop takes the same cost on each iteration because it's doing the same thing: it's loading the points from HDFS, it's parsing them into Java objects, and then it's actually doing the math. 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 each point is so little in terms of compute cycles that the time is dominated by I/O and by all the other stuff. And so if you want to use this at the full speed of your cluster, you have to keep the data in memory.
Okay. So these are kind of example applications I can put up on one slide. Spark is also being used for a bunch of real applications, both at Berkeley and at a growing number of startups actually. So I've heard from a couple of people that have started to use it. At Berkeley, one of the ones you'll hear about is the Mobile Millennium Project. This is a big project 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 of instrumentation to get. So you'll hear about that from Tim.
At Conviva — Conviva is a video distribution and analytics startup — and they've been using Spark for a while to do 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. At Berkeley we have another project called Monarch that did spam classification 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 normally I would show some slides about the applications here, but I want you to hear from the people who actually run them. So the other cool thing that we're doing with Spark is essentially building other programming models on top of it. So if you want something higher level than the Scala map, filter, that kind of stuff, you can actually implement a lot of cluster programming models efficiently using RDDs and using Spark.
So MapReduce I think you already saw — you have the map and our parallel reducer is called reduceByKey. Dryad is something that we can support as well. So Spark internally actually runs a general DAG of tasks — it doesn't just run MapReduce steps — and so you can build these kind of more general graphs that come up in say SQL queries and run them efficiently as one graph. Dryad is Microsoft's generalized MapReduce that lets you do this.
Another cool thing we can do is Pregel 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. This was actually implemented by an undergrad, Ankur, who's over there. You can ask him about it. This is open source and it's part of the Spark codebase. It's a pretty cool thing you can use.
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.
So just one or two words about the implementation before I go on and actually show you this thing in action. Spark is built on top of Mesos, which is a project that we developed earlier in our lab for managing clusters — for sharing resources between clusters — and it's also an Apache incubator project right now. So Mesos — the idea in Mesos 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 MapReduce, and we wanted to build a resource manager that can efficiently and scalably run more of these frameworks together. And that also makes it easier to build new frameworks.
So with Mesos 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. Even though Spark is built from scratch — it's not 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. And we also didn't make any changes to Scala, which is also nice because we don't want to be in that business.
Okay, how do we do the language integration? This is like one of the more interesting things, especially if you've used Scala before. So a lot of this is done by the Scala compiler. In Scala 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 on and stuff like that will get shipped.
Now in practice we actually had to do a bit of work to work around some cases where the compiler references too many variables from your closure. But the good news is we can do this at runtime without changing the Scala compiler, and it's actually pretty stable. And we have some other cool tricks like for example 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.
And then finally for the interactive Spark we modified the Scala interpreter, and 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 you hopefully a demo of this thing in action, and then you can hear about some of the applications. Okay, here we go. Yeah, let's hide this guy.
Okay. Great. So let's see. Basically for this demo, I set up an EC2 cluster with 20 machines, extra-large, and I loaded a dump of Wikipedia on it. So it's a tab-separated file. And then I set up Spark on it. So this is the Spark interpreter, and we're going to analyze this data interactively. So the Spark interpreter looks a lot like the Scala interpreter with some extra stuff at the top, and it lets you do the exact same things you can do in Scala. But the thing that you have extra is this variable sc, or SparkContext, which you use to actually access the distributed features.
I don't know if people can see that. Maybe I'll move this back 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. So I just had that in my history. So I have this text file on EC2, which is Wikipedia. And this is a data set of strings. 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 actually just take a peek at the first string by calling first.
So this is one line of text in the file and it's tab-separated. 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. 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.
Sorry about that. So let's make a class to represent the articles. Okay. So we'll have this. Okay. And now let's transform the file from a collection of strings into a collection of articles.
So what we're going to do here is we'll take each line, we'll split it by tabs, and then 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 called split. And we're going to create articles from fields number one and four.
So this is now a data set of articles. Okay. And let's try — so with each data set it can either be cached or not cached. Let's first try the not cached one with the on-disk data and see the performance of that. So let's try to run a query over this. And we'll just count how many of them contain Berkeley.
So now Spark is starting stuff on this cluster. This preferred stuff means that the block of the file is on that node, so it's getting data locality. And it's going to take a while to run this stuff. But it is running the code that we typed in. And it's going to give us back a count.
Okay. So it took 19 seconds and it says about 13,000 articles include Berkeley. Which is, you know, not that bad, I guess. So that was good, but that's not exactly interactive. That's especially once you get to a bigger file — that's at the point where you go and take a break and grab some coffee as Alexey was saying.
So let's try to also tell the system to cache the articles. And then let's try the query again. So right now the articles aren't in the cache, so it's going to take about the same amount of time. But you see when we're running it now, it's saying that it's added these entries to the cache.
