1st Spark: Karthik Thiyagarajan, Computing Time Series from Extracted Data using Spark
Recording: 1st Spark: Karthik Thiyagarajan, Computing Time Series from Extracted Data using Spark
And you know, you can do things like math or whatever that would be hard to do in Bing. Yeah, exactly. Yeah. Yeah. Cool. Okay, let's I'll give them a >> Uh hi, I'm Karthik. I'm a developer with Quantifind and I'm just going to to sort of This is a short talk about how we use Spark at Quantifind and our experiences over the last couple months. So to give a short overview about Quantifind what we basically do is we're a small startup
We're under the radar and we do predictive analytics. So basically we take data extracted from structured and unstructured information and we help our customers gain relevant insights about their past and future. I'll sort of elaborate on that. So this is a really simplistic view of how data flows through our system. So we we get feeds from various structured and unstructured information, social media, social networks, e-commerce sites and blogs. And so we sort of parse these uh data feeds into documents and then extract information out of it. So, we use a variety of algorithms like named entity recognition, sentiment analysis, uh and we also sort of extract different metadata from the documents with respect to what the dates are, uh what the geolocation is, and we aggregate all this information, slice and dice it, and create time series data out of it. And we use that to compute insights into the customer's future
Uh Yeah. Uh so, if you if you uh to elaborate a little more further into this, uh for example, the basic idea is let's say uh you have occurrences of the term screen size, which could be a key phrase found in documents, and you find that the time series data pertaining to the occurrences of screen size in the context of positive sentiment is a strong correlator to some knowledge performance indicator, let's say a revenue data from a smartphone company. And if you can sort of find a correlation between these two time series, and if you can do that with a delay, then it means that we can actually predict what the revenue of the company would be uh the delay size in advance. So, let's say there's a 1-month delay, then we can sort of say uh with a certain with a certain confidence as to what the revenue would be of that smartphone company 1 month in advance. Uh so, we work with several consumer electronics companies, consumer products, and several movie studios to do box office predictions. So, this is the basic problem. So, uh given a set of entities, I'll talk about the data model uh shortly. So, uh given a set of entities, how can we sort of apply various filters, slice and dice information to create time series data out of it? So, this is the basic uh unit of data that we deal with
This is a data record. So, this is an entity, and as you can see, it has a value and UTC. So, this could be this could be a key phrase in a document. It could be an organization. It could be a brand. Any piece of meaningful information that we extract from a document. And we make sure that in all the sort of entities that we extract, we find what the date is in the document so that we can sort of aggregate it over time to sort of compute the time series that I'm talking about. So, we use Hadoop and cascading preliminarily for our unstructured processing pipeline to run our unstructured annotators and so on
But, when it came to the problem of having these huge set of entities that we extracted from unstructured corpuses and converting them to time series, we looked at several map reduce abstractions. We looked at Cascolog and Pig. And we decided to go with Pig because it seemed to be the right idea at the time. We wanted to sort of We thought cascading in Java was pretty verbose and Pig is sort of seen a lot of success lately. There's a lot of good a lot of big deployments in big companies and so on. But, it didn't quite work out for us. So, the main problems we had with Pig could be sort of classified into two different categories. One is performance-related and one is sort of maybe development-related
So, most of the problems we ran into was development-related. We thought that the coding velocity that we got in Pig was pretty terrible. The debugging was pretty bad. So, if you write if you write any sort of Pig code that is more than 10 lines, then you're going to go into this endless loop of debugging where and you sort of have to write it in a totally different language and then put it in your grunt shell, see if it parses, and then you're dealing with data and you don't really know what the types are. So, you have to be careful about that. You could you could end up with a tuple of You could end up with a data bag of tuples of various Java objects that you don't really know what's going on. So, you have to be really careful and you have to probably use their tools to the fullest extent. They have pretty good tools now, Illustrate and so, even apart from that, we we were sort of using Pig for at least a month or so and we found out that it wasn't really working out for us
It's it's it's really unforgiving for the beginner. And so, one more problem with Pig what we had was because we were slicing data in different ways from a lowest common denominator, which is a file full of entities, we found out that if we executed more than one query, the it was it was sort of like iterative algorithm and it turned out to be really really slow. So, we didn't have sort of all the caching advantages that would be given by Spark. And the control flow, so we basically had to sort of run several of our Pig algorithms in a loop and so the the way the way you sort of achieve loops in Pig is by embedding the code as a string and then passing it to a higher-level language or sort of writing a Pig file and then sort of giving it to your Pig server and running that over and over again. That was that was a nightmare. So, and I I latency of job scheduling. So, the data sizes that we had were pretty diverse. We had anywhere between like hundreds of megs of data to 100 hundreds of gigs
And when we really when we were sort of dealing with small data sizes, a whole fraction of the time taken to run these jobs would be job scheduling rather than the actual processing itself. So, this is sort of to give you an example, simple Spark code to do one of our sort of faceting. As you can see, uh we're dealing we're reading a huge file of entities and then putting it in an RDD of and the type is entity, so it's an RDD full of entities. And then we basically what we're doing here is we're just filtering the RDD based on different properties of that particular entity. In this case, sentiment of the entity and then we're grouping by value. So, this is similar to the example wherein in the end you would get a time series, let's say of screen sizes in the context of a positive sentiment or a negative sentiment based on the parameters you pass in. So, this is another example wherein if your entity has other metadata information, let's say geolocation, and let's say it corresponds to a certain topic. We do topic modeling and decide which topic an entity pertains to
And in this case, if you were to say uh "Give me all sort of entities pertaining to topic smartphone in Europe or something like that." You could just pass it in and pass it in that way. So, because as you can see, there are several ways in which we can facet information, and it all runs on the lowest common denominator, which is an RDD of entities. So, if we can cache that and perform these faceting iteratively, then we actually get pretty good performance gains. So, uh over the course of the last couple of months, we've actually come to like Spark, and there are some pretty good things about it, especially with respect to coding velocity. Uh The good thing is the data flows are written in Scala, and so the the code in which you write your sort of data flow is the same as the code in which you actually uh write libraries and stuff. And so, it's pretty seamless, and you don't have a glue language. You don't have to write a UDF, and then write some pig code, and pass in parameters, and so on. You don't have to futz around with that
You have access to all the Java and Scala libraries. Uh type inference makes coding really easy. So, for example, if you have an ID, and then you're just writing a cascade of closures in in in in Spark, and then in the end, you don't have to worry about what the type is. You just going to know exactly what type you're going to end up with. Uh and the other one other other two are basically textbook advantages of Spark, which is because if you cache the data set in memory, then you're going to end up with pretty low latency. And finally, the interactivity of Spark opens up other possibilities that I'll talk about. So, one one one problem we had in Quantifind was uh it's fine that we had an analytics pipeline. We know exactly what set of queries to run and extract time series out of
And so we had this sort of pretty deterministic pipeline. We sort of fascinated it by a deterministic set of parameters and then got the output that we wanted. But we also sort of use other languages to prototype our code in. For example, we do a lot of statistical modeling in R and then so when we wanted to sort of we wanted a way in which people using R or the data scientist could sort of query our data interactively and sort of use that in their modeling. And so one way to do that is a Spark shell. But the problem with the Spark shell right now is that only one person can connect to a cached RDD at a certain point of time. So if there was a sort of more or less deterministic way in which you could several people could access the Spark RDD at the same time and we could expose a service that would be awesome. So So what we did was we used Spark and Unfiltered
Unfiltered if you guys have not heard of it is a HTTP toolkit that basically is a really lightweight framework that runs a Jetty or an Eddy server and then routes HTTP requests based on regular expression matching. In this case what we do is we've cached the RDD again in memory and then we're listening for HTTP requests at that particular server. And you can actually pass in whatever parameters you want. You can say entity.value equals so and so and give me all the time series for this particular sentiment. And this would actually compute it on the fly when the server is running and spit out the output you want. It's It's It's also pretty good because you can also update the data when the service is running and you don't have to take it down. Let's say you're running a service after you cached hundreds of gigs of information and you want to sort of update it with 1 gig of updates that you're getting on a given day or something. Uh you could actually sort of send a post request, and then uh union your new updates to the cached data while in memory while other people are using it
It would block, but it would actually sort of be faster than reloading the entire RDD. So, through this experiment, simple insights that we've gained is that uh you could really use this as a very good sharded distributed service. Uh the distribution of data is uh abstracted from the application. Spark does it at a at a lower level, which is good. Uh caching guarantees low latency, and so it's almost real time. Uh we've sort of had gigabytes of data in memory, and then query times of less than a second, which is actually pretty good. Uh fault tolerance is built in, but then if a node goes down, the the HTTP request handler is going to block because Spark is going to sort of uh get that node up before moving on to further requests. And data update is possible
Uh what I did What I've shown here may not be the most elegant way, because then your cached data would have to be a var, and then you'd probably need to reassign the variable there. But that's just to sort of give you a picture of how it looks like. Uh the bad is that replication would have to be implemented by the application. Uh because if if you have a if you have a lot of data load, and there are several people hitting hitting it at the same time, then you're going to have you're going to block till Spark finishes its computation. And so you would probably have to deploy several instances of it, and that would be a that would have to be managed by your application. And if you have queries that are highly constrained, then it works out pretty good. But if you have queries in which uh the data you return is huge, or for that matter you have queries in which there are a lot of shuffle operations and group by operations involved, then you're basically creating uh data in memory that is sort of almost equal to the size of your cached RDD. And so, you would have to probably rely on the garbage collection to sort of clean it up before your subsequent query set
So, you'll probably have to sort of look at that. >> What size of data are you talking about? >> Excuse me? >> What size of data are you talking about to give us >> Yeah, so we've used it with gigabytes of data right now. Uh we're trying to sort of get it to work with hundreds of gigs. But we have So, we've just sort of trying to put up a cluster right now uh so that we can get the service up at the order of hundreds of gigs. But if your data is if you if your queries are highly constrained, let's say even if you have hundreds of gigs of data, uh the answer that you're expecting is several megs, it should still be fine. But let's say that you're applying a map operation or something. Um let's say you're stemming all the entity values or something. Um so, you would actually create another RDD which is of the same size
That may not be cached, but you're going to wait for the garbage collection to do it. >> And what size of cluster and what's the limit you can work with? >> Uh so, you can So, technically there is no limit because it would spill onto the disk. >> I mean, nodes in Spark. >> So, we we've run on clusters of up to like 200 external RDDs to nodes. So, 8 core each. Yeah. >> Yeah, so uh that's that's basically the end. Uh if you guys are interested in what I was talking about, here's a sort of shameless link to our uh jobs email
That's it. >> Are your slides going to be available somewhere? >> Yeah, I'll put it up. I'll put it up in the meetup. >> Can I give you an example of what kind of transformation? So, you start at series from text data. Like what kind of uh extraction is going on >> So, you're talking about >> You're you're you're kind of playing series. I was just curious what kind of So, in order to get to that, what is the source data and how like what's happening inside? >> Let's just say that you have a bunch of unstructured documents. Right? So, you could do something like named entity recognition on them. And so, what you could say is, "Okay, like give me all the organizations in the document." And let's say that these organizations put into certain sentences uh from which you could extract sentiment for for for that matter, right? So, let's say Apple is mentioned in a sentence with positive sentiment in a certain document in a certain site, right? So, you can actually aggregate this across all pages, across all sites, and then combine them together to give you a time series
>> Okay. So, you basically you run like Scala and Spark on your machine or is that the named entity recognition on the document or just regular Scala and Spark, right? So, and then you're going to be extracting that. >> Yeah, so right now we we use Hadoop for our information extraction. Uh so, until the point where we actually have a huge list of extracted entities, it's all Java and Hadoop. So, Scala comes in Scala and Spark comes in in our analytics pipeline where we want to sort of slice and dice these entities. >> Okay. So, you have already time series extracted. You just want to zoom in, you want to do methods on the data
>> Yes, exactly. So, so right now the maximum cluster size we run on is about four or five nodes with about 48 gigs each. Thanks. >> How did you run Did you find situations where the actual things lost their IDs, things, you know, because they run out of reading in the data. What's your practical experience >> Uh we've had problems with it initially wherein when we sort of try to run it on a cluster, um we've had those angs, and then it used to lose tasks and all that, but uh we've actually sort of reported the bugs, and it's been fixed pretty uh periodically. >> What's the way to test the live as part of Java? >> Yeah, so so to test the Java running in local mode, just pass local as the master. It's the easiest way. It will be in one JVM
You can use your standard Java debugger and add breakpoints and stuff. >> Do I have to restart between each test? >> Um you can use Yeah, you can start a new Spark context for each test and then close it. So um in the same Java program, you can have you can use Spark context multiple times. >> That isn't going to be quick enough. >> Yeah, it is. It is. Locally, just starting a local is very quick. It just makes a new thread
So all our unit tests create a new Spark context. >> Thank you. >> Let me Let me just put up one last slide. Any more questions? I don't think I'll need the mic. >> Um Yeah. >> All right. So yeah, so just just to finish up. I also wanted to say so so we want to do this meetup regularly
We're thinking every 3 weeks or so. It may change. And for the next meetup from Viva, which is a startup in San Mateo that does streaming video and a lot of analytics on said video, uh volunteered to host it. So we're going to to alternate between different locations. We're hoping to have some in the South Bay as well, and we'll definitely come back to San Francisco, too. Um and we'll see which which way ends up working well, cuz Viva is sort of in the middle. Um and one other thing, um we're really curious to hear from you. So we put up this little Google Forms here, where you can give us some feedback
Tell us what kind of things you're interested in hearing, uh what kind of things you'd like to see in Spark and so on. And if anyone has um topics they want to talk about, that would be great. We have a bunch of topics that we have lined up and we have a bunch of companies that want to talk about what they're doing. But any specific, you know, like give me a tutorial on how to do PageRank or how to do um K-means or logistic regression or something like that, we can do that. So, thanks. Thanks a lot for coming. >> [applause] >> What are you guys doing with the streaming? >> What's doing? Yeah, so the idea >> I mean, I've been looking at the storm past couple of >> Oh, okay. Yeah, so the idea is um