About two years ago, this project went from being sort of this research prototype to being actually in production use at a bunch of places. Here are some of the users on here, and they're using it for things ranging from just replacing Hive and Pig to do rollup kind of analytics, to developing new machine learning algorithms, or even to doing live queries on sort of streaming data — so loading new data and letting users query it interactively.
So here's kind of the programming model. The main idea in Spark, similar to the other frameworks you've seen, is a kind of distributed collection. We call them resilient distributed data sets, or RDDs. And the resilient is because they're also fault tolerant. These are distributed collections of any Scala object you want. And the one thing you can do in Spark that you can't do with just Hadoop is you can also tell the system to cache one of the RDDs in memory across the cluster because you're going to use it in many parallel ways. And other than that, you manipulate them the same way you do with local collections, and Spark also knows how to rebuild them on failure by re-executing tasks, much like MapReduce.
So let me just jump into an example of what you might do. This is how you might use Spark interactively to explore some big data set of logs. So say you have a bunch of logs, and it happens very rarely, but sometimes people make a mistake in coding an application and it's crashing and you get all these errors. So here's what you do. You launch a Spark cluster. You have a master which is where you're going to be typing in commands and a bunch of workers. And you type in the following Scala code at the shell. You would first start by making an RDD called lines by loading a text file in HDFS, and this is your base collection. It's a bunch of strings. And then you can do operations like this: filter it, pick out for example the lines in the log that start with "error." And I'm going to just breeze through this because you've seen this kind of stuff before. Do a map to split them and to pull out field number two. Maybe it's a tab separated format and you want field number two. And finally, at the end, you can call cache. You can tell the system, okay, load just these error messages into memory and keep them around because I'm going to ask a lot of queries about them. So once you do this, nothing's happened on the cluster yet. All these transformations of data sets are lazy. But you can also launch another type of operation called an action, such as count. This is an operation that has to return a result. Count is give me, you know, how many of them match this filter. So it's going to return a number. And when you do that, the system is going to look at where the data is in HDFS and it's going to farm out tasks, and everyone's going to compute their local count and ship it back. And it does all the smart things like pipelining the maps and doing combiners and stuff like that. The other thing it's going to do is it's going to build a cache of the data set partitions for the error messages that it's built along the way. So now next time you run a query, you know, type a different query in the Scala shell, you want to say, okay, Foo was the problem. Let's search for Bar. Now the scheduler will know the data is in the cache and you'll hit the cache and you'll get back the results much faster. And of course, you don't just have to do filter and count. You can do all kinds of other operations.
So what difference does this in-memory caching make? Just as an example, one of the tests we did was just doing a full text scan of Wikipedia, which is about 50 gigabytes of text. And if you do this with Hadoop or with on-disk data in Spark, it takes about 20 seconds. And if you load the data into memory and then do the same scan, it takes about half a second. So it really makes a difference to have the data sitting in memory as Java objects because you avoid a lot of the overhead of doing IO and even just deserializing, copying bytes around to get a string. And just for fun, we also scaled this up and, you know, we tried a terabyte of data this time out of 100 machines, and we could do that in about 5 seconds. And if you've used Hadoop and tried to use it interactively, the reason we've done this is because we saw everyone who sets up Hadoop tries to ask queries interactively, maybe with Hive or Pig. And even a MapReduce job that does nothing takes 20 seconds just because it's a batch processing framework. And with this, you can actually get a speed that you can call interactive.
I'll give another example. So this is the other type of thing you might want to do with Spark where the in-memory caching really helps. This is a machine learning algorithm, logistic regression. And the idea here is basically it's an algorithm that's a bunch of small MapReduce steps that all share the same data. The goal of the program is to do something like classify spam and non-spam. So you have labeled objects in some feature space, some high number of dimensions per point, and you want to find a line that separates the two sets of objects, the spam and non-spam. And the way a lot of these algorithms work is essentially gradient descent. They start with a random line at the beginning, and then they compute this gradient function which is a sum over all the points. You can view it as each point pushing or pulling the line in some direction. And that gradient function gives you a direction to move the line in to improve the separation. And you keep doing this and eventually you converge to a pretty good line. This is one of the standard things you can do to solve these kinds of problems. So what's interesting about this is each parallel sum is a MapReduce but they all share the same data. So if you can keep that data in memory you'll go a lot faster.
So you can write this algorithm in Spark as well. Here's again some Scala code for it. And you can see it's very simple. You start similar to before. You take a text file, map it, maybe you turn the lines of text into point objects that you designed the class for, and you cache those objects. And then you have a loop and you call map and reduce. And don't mind the math if you haven't looked at logistic regression before, but that's the math for it. One thing that's kind of cool is the variable. So w is the current separating line and you use it in the function and it gets automatically shipped to the cluster. It's a nice thing to be able to do. So this is the code, and this is the comparison with Hadoop. We tried the same thing and implemented it in Hadoop. So what we see here — this is on a 20 node cluster — as you increase the number of iterations, on MapReduce jobs Hadoop takes the same time per iteration, which is about 2 minutes. And Spark on the other hand does the first iteration in maybe three minutes, but after that each other iteration is only six seconds. So once you load the data in memory, again, you can go a lot faster. And the reason why is again it's avoiding this IO and it's avoiding serialization overhead from having to write the data out to HDFS or load it back in on each job. And so I think this is — if you take away anything from this talk, I think this is one of the most interesting things we found. It's that when you write this kind of application in Hadoop, it will work well, it will scale linearly as you add more nodes, everything is great, but the program is spending 90% of its cycles doing IO and doing serialization. And if you instead keep Java objects in memory on the machines, you can go significantly faster. And you can actually match the speed of what you'd get if you just wrote a single-threaded program on your machine and looked at the speed of the CPU. So we've had a lot of users tell us that, oh, we started with a Pig job and then we implemented it in Java and actually it was faster on a single machine than on our cluster, and then we implemented it in Spark and it was even faster. So it really helps to get a good baseline for the performance you should expect from your job.
So I've showed some simple examples with map and reduce. We have a whole bunch of other operators too. I'm not going to go into them right now due to sort of lack of time, but we try to add new operators as people find them useful. Let me talk a little bit also about some of the other things we do that having our own execution engine, not MapReduce, gives us the advantage of being able to do. So one of the things we do — I don't have a slide for this — but we execute general graphs of tasks, not just map and reduce. And what that means is if you have something like say map and reduce and another reduce, you don't have to save the data to HDFS between each map step, and you can actually often go faster even if you're not caching anything just by virtue of that. That's one of the things, and all these operators support being built into a graph like that. Another cool thing we do is we also allow you to control the partitioning of intermediate data sets.
So this is a thing. If you've ever written PageRank, for example, you know that in PageRank you're joining a bunch — you have like a list of links for each page and you also have its current rank. And on each iteration, if you do it with MapReduce, you have to do a big join where you shuffle around the links and the ranks to get the link and rank for the same machine into the same reducer task, and that's a lot of communication. If you can instead keep the links partitioned by page, you can cut down that communication. So we support — we let you control the partitioning of these data sets to do that. And the final thing that's interesting — this is probably the technically hardest part of implementing Spark because there's a lot of sort of hacking involved — is we support using this from the Scala interpreter. This is non-trivial because in the interpreter you want to be able to do things like define variables and functions on one line and then use them in a parallel operation on the cluster and have that value automatically be shipped to all the nodes. And we essentially modified the way the interpreter generates Scala code for each line you type in, and we modified the class loaders and the remote nodes to be able to load those objects that you defined before. But the end result is that you can use these features here in the interpreter as well.
So just to sort of wrap up, I wanted to show you that in action, and we'll see if the internet works, just so you can get an idea of what you can do with this stuff. And then I have links and other things if you want to hear more about the actual project and how to use it.
So I'll try again to shout like other people have so that I don't make weird noises with the mic. Okay, so what I've done here, I've set up a 20 node cluster on EC2 using Spark, and I've set up HDFS, and I've loaded this 50 GB dump of Wikipedia. And I'm just going to do a short demo where we do some queries on Wikipedia interactively. So I have a screen for that. This here is a Spark interpreter. Looks a lot like the Scala interpreter with some extra stuff at the top, and you can do all your standard Scala stuff in it — like, you know, do some math, define some variables, and so on. The other thing you get though is you get this special variable sc, or SparkContext, that lets you run jobs on the cluster. This is the same way you run jobs in a standalone program too. And so from sc you can do different things. The first thing we're going to do is create a data set representing a file in HDFS.
And I have that typed here just because I don't want to memorize this domain. So we created this file for Wikipedia, and it's a distributed data set of strings. So as soon as you create it, you can start sort of exploring the data. So for example, you can call file.first to get the first element of it, the first thing. And the scheduler is smart and says, "Okay, I'm just going to open the file and read the first line." It doesn't launch a whole job. So this is the first line in the file. And just to give you a sense of what's in this dump, this is a tab separated values dump of Wikipedia. And each article has five fields. It has an article ID, it has a title — this was maybe the alphabetically first data action pack — it has a date modified, it has an XML version of the article, and you can see the last field, but that's a plain text version of the article as well.
So what are we going to do with this data? So first of all, let's define a class to make it easy to work with this so you don't have to deal with tab separated files all the time. So you can define a class Article, and I'm going to pull out just two fields from each article — the title and the... Okay. So I defined this class, and now I'm going to turn this data set of strings into a data set of Article objects. So let's see. I'm going to take the file and I'm going to map. I'm going to split each line by tab. And it turns out some of these articles don't have — some of them only have four fields. They don't have the last field which is the plain text because there's something like an image. So I'm going to take only the ones that have five fields. And now I have the five tab separated fields. And I'm going to take this array and do a new Article and take field one and field four.
Okay, so now I have an RDD of articles. And again, this is computed lazily so nothing's actually happened. And if I want, I can do something like articles.title and you can see it's active. So let's start searching. Actually the last thing I'm going to do here, I'm going to tell the system to cache articles after the first time it completes it. So I just told it. Finally, let's start doing some queries on it. So I'm going to do it this way. I'm going to count how many of these contain — and let's run it. So first the data is not in the cache. So this is going to run a job that reads from disk, very similar to running a Hadoop job, and the speed is similar to running a Hadoop job. So you can see it's running all these tasks, you know, some of them are finishing. And it says "cache entry added" because it knows that it's cached a partition of the data set. And it's kind of cool that we were able to launch this job from the interpreter, but this is not exactly interactive. I wouldn't call this interactive exploring the data. But it's going along, and once it's loaded it into memory, will be very — yeah, there we go.
So that took about 30 seconds, and the result — the number of articles containing "Berkeley" — is about 13,000. So let's try that again now that the articles are in the cache. Sorry, same thing. And then this one actually seemed to take longer because of the network, but this one took about 0.9 seconds. And you know the result is the same. So this is really the difference with working with in-memory data, and this is the kind of thing we want to enable. We want to enable people to use the Spark shell as an extended sort of Scala interpreter to do interactive analysis on large data sets. And of course, once you have this stuff, you can ask other queries too. I'm just going to ask one more for lack of time. So Berkeley was 13,000. Let's look at Stanford. And Stanford is only 11. This is the second major... So yeah, so that's a very quick introduction to Spark.
If you want to get started, check out spark-project.org. We have a download, we have documentation, and you can run the program very easily. If you want to run in local mode on multiple cores, you just need Scala installed. And if you want to run on EC2, we provide scripts to do that. And if you want to deploy to a cluster, you can currently do it with Mesos, which is a cluster manager we developed. Soon you'll be able to do it with YARN. And finally, we do have a regular Spark users meetup. The next one will be in Palo Alto on April 23rd. So I encourage you to sign up at that page and watch for these meetups if you'd like to hear more about it. Thanks.
All right. So you can run it, so it will read from HDFS. If you want to run it alongside, say, MapReduce or something like that, you can do it either by running it in a separate cluster or using Mesos. We actually support running Hadoop and Spark on the same cluster and sharing resources between them in a smart way. But it will work. It's possible to use it on both, and most people use it alongside. ... So it's possible to use it standalone. And the way we recommend deploying it, if you want to deploy it in your own cluster, is using Mesos. If you want to try it on EC2, the scripts that we have actually launch a Mesos cluster for you and they do all the setup. And it's really the easiest way because otherwise you'd have to install Mesos and Spark on each node yourself, which is a bit of work. But it's a pretty easy process. So you can check that out. The scripts are also on our website. Okay, thanks.
