mid-2013, my Spark Odyssey... when Akka sparks Yahoo! finance

This entry is the second of a series that relates some work/test I'm doing using Spark. All information can be found in the very first post of this series.

In the previous post, I've explain how easy it is to consume a Twitter stream using Spark, and seamlessly easy how to attach a sentiment score to each tweet passing a given condition (read: filtered).

In this one, I'm going to tackle another fold of my test, which is fetching the current stock information for a given list of companies.
These information will be provided by the Yahoo finance API...

...well, hum, not exactly an API. Actually, Thanks to @kennyhelsens who shown me this "hack", we'll use a Yahoo! URL that produces a CSV file containing real-time (mostly) information about companies' stock changes. Information are really hard to grasp for this URL's parameter however you can find some out there.

Storyline

In this series, we're trying to uses real-time streams rather that batch processing to give a real-time fluctuation of the health of given companies.

But we've just seen that we're going to consume CSVs that give a snapshot of the stocks changes.

A common solution to transform a static API to a stream is to poll it and to create a streams by appending the results one after the other.

Here, we'll try a pretty simple solution for that, we'll delegate to Akka the task to consume the CSV, to parse it, to split data by company, and to modelize each and finally to push everything to a DStream.

It's rather straightforward and we'll see that it's even simpler that we would expect since, Spark is already playing a good game with Akka by integrating out of the box an API to "consume" actor's messages.

Acting

In this scene, there will be several actors each playing its own roles (sorry Peter Seller's fans -- like me btw).

Here they are:

• scheduler: this one is provided by Akka out-of-the-box and is able to send a "given message" to a given actor at a given interval.
• feeder: this actor will simply fetch the CSV and produce the data to be sent to the below actor.
• receiver: this one will receive a message whenever a data is available for a given company.
• parentReceiver: this one will be created by Spark itself using the StreamingContext and is able to push the received Data to the right RDD in its holding DStream (see ActorReceiver).

Now the scenario:

Sequence from the scheduler to the DStream (current RDD)

Does that make sense? Let's have a short overview on the implementation.

Scheduling the poller

First we have to create a scheduler that will poll every 500ms the server... well using Akka and the separation of concerns and Message Passing Style, we'll just create a scheduled action that passes a message to a given actor:

But even before that, let's prepare the Akka field:

There are a plenty way of creating ActorSystem with Akka, but Spark is preparing some stuffs for us to create one that will respect some of its own internal conventions. This helper is available in the `spark` package, naming AkkaUtils; however it's package protected... That's why I've created my own utils to create the ActorSystem under the spark package -- I'm bad sometime, but I'm so often lazy :-D.

Actually, the Spark utility for Akka will create an ActorSystem using a host and a port for us, that'll be really helpful afterwards... because everything in Spark must be serializable and thus we cannot so easily package an ActorRef (needs an ActorSystem, ...). So that, we'll use URL to create refs rather than using actors picked within a closure in a Spark action.

Recall: that's how Spark manage to recover from a failure, it's take to lineage of the RDD or rebuild everything up from the initial data (not possible in Streams...). But specially, the actions (functions) are dispatched to machine where data reside. So the lineage contains the functions, and a function might contain ActorRefs!

Now that we have an actorSystem available, let's create the scheduler (see last section, Drawbacks, for more info about problems with that methodology):

Nothing relevant to say about that, only maybe the definition of FeederActor...

Feeding Spark with Yahoo data

Here is coming the feeding phase, we won't see how the CSV is read and parsed however we'll see what going on with the messages flow.

What's interesting in this actor is two-folds:

1. it can receive a message containing two data: 
1. an actorRef that it will hold in its state: this one will be the receiver
2. a list of stocks to follow (by building the according call to Yahoo)
2. when consuming the data and producing the data by stock, it sends them one by one to the above actor

Actually, this feeder has two responsibilities which are fetching the data and pushing it to Spark.

But how is this receiver constructed and passed?

Receiver -- our spark stuff

The receiver is actually an actor that will be created internally by Spark for us, for that it must respect yet another trait, Receiver. And a Receiver can only be an Actor (using self definition)!

Four things to point here:

1. this actor is well-defined by extending the Receiver trait, which provide the pushBlock method.
2. it holds a reference to actor that it, it-self, create based on a give URL -- an actor URL... this is where AkkaUtils came handy! This Actor will be the feeder actor.
3. in the hook preStart we tell the feeder actor that this ActorRef is the Spark end-point where data must be sent to publish in the stream.
4. when data arrives, we check in the cache if it's already there (see below), if not we call the Receiver#pushBlock with it.

Why do we have to cache? Actually when the stock market has closed, Yahoo will continually return the last change before closing... so, here, we just avoid telling our health checker that the mood is monotonously increasing or decreasing (or even just flat if zero).

The pushBlock method is the clue to publish to the underlying DStream managed by Spark, actually it will wrap our data into an Internal representation before telling the parent actor to use it: context.parent ! Data(y).

But what is that context.parent?

parentReceiver -- not our spark stuff

This receiver is totally managed by Spark, and so I won't dig into more details than just saying it's being created underneath when doing the following:

Fairly simple, yes, it's true! Actually, the StreamingContext we created at the very start of the project contains everything to create DStream based on Akka, all it needs is a reference to a Receiver (defined in the previous section).

What is this actorStream doing is creating an ActorReceiver (with a Supervisor and blahblah) which will itself create and manage a worker for the given Props, but also it will hold the reference to the current RDD in the DStream.

The worker will be able to send him messages of the form Data(d) using pushBlock, and which it'll push to the DStream.

Concise huh? If only the document was telling us that, I wouldn't have to dig that deeper in the code... the good side-effect is that now I understand it pretty well. 

So, I hope you too!

Where are we?

Up to here, we've created a DStream of data for each tweet paired with 'its' company, but also we have now some info about what the stock change of them.

What will happen in the next blog is that we'll create a multiplexed stream for which we'll compute a value for every window of 60'.

Drawbacks of the/my methodology

At this level, I can see two problems.

The scheduler

Using a scheduler is bad because we cannot ensure that each message are processed sequentially by Spark. So it would say that we push data to an RDD that correspond to the time where the whole CSV has been processed but also the messages sent by the actors!

I put this aside for the moment since I'm not building a real-life project (no-one will use this tool to take decision on the marker ^^).
Moreover, mathematically, let say at the limit, the value will be correct: reducing an amount by 2 at t⁽¹⁾ or at t⁽²⁾ is not really problematic, it's like inverting two ints in a list that we're summing over.

The consumer

I didn't shown its current implementation because its very simple and bad, that is I'm simply opening a connection on the URL, consuming everything until the end before applying the process...

That says: it's blocking and non-reactive.
Also, That says that we loose the power of parallelism since the Akka message dispatcher will have to wait a CSV to be completely processed before handling the next one.

However it's just an implementation detail that can be easily worked around without modifying the solution.

mid-2013, my Spark Odyssey...

Short intro

In this post, I'll talk about yet another technology-tester project I've worked on these last days: Spark.

Long story short, spark is a handy distributed tool that helps in manipulating Big Data from several sources -- even hot streams.

TL;DR

In the first part, I'll put the context in place.

1. I'll start with a gentle bunch of words about Spark and Spark-streaming.
2. Then I'll give some bits about the project I've realized so far, and why.

The second part will discuss some implementation details.

1. Setup: spark streaming
2. Twitter stream filtered by keywords to DStream
3. Tweets' sentiment analyzer
4. Yahoo finance CSV data and the actor consumer
5. Actor consumer publishing on a DStream (back-ended by a dedicated actor)
6. Aggregation (union) based on a 60 seconds length window
7. Publishing the results using Spray

What're missing so far (among other business related things for instance):

1. cluster deployment...
2. ... and configuration
3. probably a storing strategy for further consumptions (Riak, ...?)
4. a web application that shows the results or configure the processes

Context

Spark

I won't discuss Spark that much because there are already a plenty of good blogs/articles on that. However, I'll give you just the intuition that you got the whole thing about it:

• Spark is a collection of functions that works on a Sequence (List) of typed data  --  Dataset = Sequence ; operation = Function
• Chaining these functions will define a workflow for your data  --  combination
• Each data will be processed only once by one or another Spark worker  --  Distributed
• If it fails for some external reason, the data can be fetched again from the source and the workflow to be replayed entirely  --  Resilient

Spark defines an ResilientDistributedDataset which is very different that the classical MR -- because it is distributed by essence, a workflow can easily be iterative and doesn't require intermediate caching/storage.

Also, spark as a sub-project spark-streaming that allows manipulating streamed data. The idea is very simple by defining a DStream as being a sequence of RDDs containing the data for a certain slice of time.

Oh, RDD and DStream are typesafe -- the whole thing being Scala code.

The spark-streaming project is really well integrated and can consume a lot of streams like Twitter, File System, Kafka or even an Akka actor.

Project

This blog relates a small project publicly available on github (spark-bd) that helped me catch the big picture of Spark, after having followed the mini course here.

The project is based on another working group I'm participating in. This project, simply called project 2, has been initiated and is still supported by the great Belgian Big Data user group (http://bigdata.be/).

The idea was to co-create in several workshops an application that given a list of companies to analyse;

• catches the Twitter stream and the Yahoo finance data using Storm;
• manipulates both by either applying a sentiment analysis or simply the changes in price;
• aggregates by window of time both information for each company
• gives (constantly) the information about how is going a given list of companies on the market.

Fairly simple and really funny... the project is still ongoing but you can check its progression here.

It was really funny but I wanted to see how my feeling would compare by writing the same application using my actual preferred language (Scala) and the next technology in the Big Data field I was eagerly looking at (Spark).

IMPORTANT: since if I'm not there yet to make a real comparison (a niche for another blog, later on), I'll only give an insight on my actual work.

Internals

Setup

The setup is very minimal for a Spark application. All we certainly need is reduced to two things:

1. an SBT project declaring as dependencies Spark and Spark-streaming.
2. in a main class/object configure the spark streaming context.

The SBT project only requires an build.sbt in the root folder that contains these lines only:

Where the only lines that are under interest are the two last ones.

Then we can create an object that will start the spark context.

Here we just asked Spark to create a streaming context on the local environment, named Project2 and that will slice the DStream into RDDs of 5 seconds duration each.

Afterwards, we asked the context to start its work... which is nothing up to now, since nothing as been added yet to the context.
So let's do it.

Twitter stream

Here we'll consume the tweets stream provided by the Twitter's API. For that, we'll have to do only two things (a pattern emerge...?):

1. register the 4 needed information for an OAuth2 authentication to the Twitter API (so don't forget to add an application to your Twitter development account).
2. add a tweets stream to the spark context

Actually, the streaming context as convenient methods to add a twitter stream.

For that it'll use the Twitter4J library (it's definitively not the best, but at least we can achieve our goal).

This library needs several keys to be set in the System.properties.

There are a plenty of way to do that, in my project I took something rather straightforward: I've picked the Typesafe config library and added those four configuration keys in an application.conf file, then I load all value in the related System property.

As you may (or not) see, the application.conf file will look for environment variables giving values for the required key/secret/token.

That's done we can start listening for tweets:

Dead simple, after having created a model for a company/stock, we used the configured keywords as tags for the twitter stream.

Behind the sea, Twitter4J will access the stream end-point with these tags, and Twitter to pre-filter all tweets based on those keywords (in hashtag, username, text, ...).

At this stage we've only added the stream, but we didn't yet said anything about how to consume them... So we can simply call print() on the stream to ask spark printing the 10 first incoming event in each RDD.

However, that's not how we want to deal with those tweets right? Let's adapt our stream to add relevant information for our use case.

Sentiment analyzer

Now that we have a stream of tweets it's time to read them and to give them a sentiment score based on a sentiments file. The code that reads it and that assign a score for a tweet is not that important here, but the code is here.

However, what we would like to do is to adapt our stream with this score attached to each tweet.

As said earlier, Spark is a bunch of methods on Sequences of data, so we can now assert that it's true. 

Look at what was done in the map, we picked each tweet, fetch the list of sentiment entries that match the text, and also we kept the whole original status for further computation. This results in a DStream[(List[Sentiment], Status)].

Since we only want those tweets that have at least one Sentiment that matches, we then filtered on the length of this matching list.

We combined both actions to get a resulting filtered and sentiment-analyzed tweets.

Okay great, but what we really want is to have such information grouped by Company/Stock right? So that we can compute the participation of the social network to the health of each company.

To do so, we have to group semantically (based on text's content and required stocks' keywords).

Note that it's not that straightforward because a tweet could relate several companies at once!

That's all! We're now done with the twitter part, we are now getting all Data that gathered the information of a Company/Stock, the tweet and its associated sentiment. Which opens doors for further aggregations/grouping/reducing.

Interlude

Let's now pause a bit, I encourage you to try your own stuffs and playing with the tweets.

For that, I also encourage you to fork/clone the original repo, where you can already start only the twitter feed and print everything in the console by running the appropriate command:

$> sbt

sbt> run-main be.bigdata.p2.P2 twitter print GOOG AAPL

This will run the start spark and filter the tweets based on the keywords associated to both GOOG and AAPL, whose tweets will be analyzed as said in the last section before being printed to the console.

WARN: don't forget to export the required environment variables

Next

In the next blog posts, I'll continue with the explanation on how use Akka to construct Spark streams and how to combine several streams into a single one.

This multiplexed stream will then be used to compute an aggregated data by a window of time.

Having these data in hand, we'll be able to create a Spray server that expose them into a JSON format, helping us creating a client web application that simply consumes the JSON messages.

Still Playing... but new players are in

Why?

Because I love to play, with Play 2.0 and scala (still learning).
But also because I'm currently investigating technologies that I might choose for a new product line currently building in my company, NextLab (in Belgium).

With ?

This time, I've more played with client side libraries or frameworks (no I won't post yet another entry on JQuery...), but I also tried how it is easy to create totally async code (so parallelized) using server side ones.
The technologies that will be quickly introduced in this post can be found hereafter, but everything has been packaged in a github repo, and a running instance on Heroku.

At first the current spike was dedicated to the slowly growing Play 2.0's Neo4J REST plugin, we are creating at NextLab. But, in order to demonstrate what could be done with the coupling of these two technologies, I've extended the spike's scope to something more funny.

So let's introduce the technologies:

Client Side

Twitter Bootstrap

An amazing toolbelt helping building responsive website without having to bother boilerplates in HTML, CSS.

Even if it is neat, complete and well thought, Bootstrap comes with another handy factor... it has been built using LESS. And by chance (I know, chance is not part of the equation) LESS is supported by Play 2.0 by default.

Just a note, LESS will let you reuse color codes, mixins, etc etc that Bootstrap has already defined.

Spine.js

As we'll below, we'll have to discuss with WS (json), upon which a REST interface has been added for meaningful resources.

That's where Spine.js comes in the game. This lightweight MVC library brought me the small tools that I needed for fetching, saving resources without having to write not even a single request by hand...

d3.js

Probably my favorite (that might be my mathematician part who's talking), this powerful Data-Drive Document toolkit has taken the right thing by the right end.

Its functional approach of decoupling data from the document, and link them using layouts helps you to concentrate on each part of the data usage independently:

• the incoming/rought data mapping to a representative data
• the mapping from represented data to document (most of the time, one data for one element)

Communication

The communication layer is of course HTTP, with a little help from the emerging HTML5 features. One in particular, Server-Sent Events (here is a great intro).

This, stable, HTML5 feature comes with the handy functionality to let the server sent events to connected client, without hacks; that said Comet or Polling.

Open a connection, push data, and that ONLY when the server needs to.

Server Side

Play 2.0 (Scala)

Of course... But I used some "advanced" feature like,

Async

Async is a Play 2.0 feature that let the server deal with the tasks it has to schedule.

That is, when you think that the server might have to wait for actions being executed before being able to respond a request, Play 2.0 let you, really simply, create Async request (non blocking).

Very handy when you have to call third parties services for instance...

Iteratee

The only way I would consider from now to consume data. Iteratee is a fairly difficult thing to understand (read this wiki) but it gives you the same smart decomposition that in d3.js, that is, decoupling the management of a sources, its useful representation and its computed result.

Akka

Powerful, actor-based, parallelizer, asynchronous task, scheduler library.

Actually I needed, a request to launch async tasks (you know like event generation and dispatching)... so what else!

Neo4J

A database for storing graph... let's use a graph database.

Within NextLab, we started a open sourced Play2.0 plugin for calling the REST api provided by a Neo4J Server (helpful on Heroku). It's still emerging, and continue to evolve a lot because features are implemented on the fly (need), and a re-pass is forecasted to add a meaningful DSL (like FaKod did).

Why do we make this choice to implement a distinct plugin instead of wrapping the java library?

1. yet another library, which brings me too much (I need REST only)
2. I want requests being async and under control

Libraries Repo

See below, we'll use Heroku. So in order to deploy this application wich uses our plugin, I needed to publish somewhere.

This is what can offer Cloudbees. Among other great things like git repo, CI and so on, Cloudbees provides you for free four maven repositories that you can make public if you wish.

So I used sbt to publish on my "snapshot" repo on cloudbees, letting heroky has access to it for downloading the plugin dependency.

Platform

Free, reliable, easy to use with scala and Play 2.0... which else than Heroku?

So what

I think that this post is already too long... However, I can let you play with the resulting app here.

Check out the code (and fork it) there.

Depending on the comment, I will expand this post to other ones to respond potential demands (if there is any ^^).

Good play.

Oh yes, one last note, the application is to Create Stuff which contains dummies. Stuff are created using a simple form that must be fulfilled. Stuffs can be linked one to another by clicking the graph.

Please create Stufs and links, it will be a good test for Neo4J, the plugin and Play 2.0.

Hope you've reach this.

SMAK. hehe