Spark + Cassandra: The basics + connecting to Cassandra from spark-shell

A lot of people are getting excited about Apache Spark. The release of the open source Cassandra connector makes a technology like Spark even more accessible. Previously to get going you'd need a Hadoop infrastructure, now you can do away with all that and start using Spark directly against Cassandra, no HDFS required.

My last two posts on the topic were all about setting up a vagrant VM with Cassandra and Spark installed. That's all well and good if you already working in the JVM ecosystem, you know what vagrant and Ansible are and you love your self a bit of SBT, but now it is time to take a step back. This post is aimed at getting you started with Spark and Cassandra without the assumption you know what sbt assembly means! By the end of this one the goal is to be able to execute (and understand what is going on) Spark/Cassandra jobs in the Spark REPL, and the next article will be submitting a standalone job.

I'll assume you have Cassandra installed and that you have downloaded a Spark bundle from their website. It doesn't matter which version of Hadoop it has been built against as we're not going to use Hadoop. If you don't have a locally running Cassandra instance I suggest you just use the VM from the previous article, follow this article for Ubuntu, use homebrew if you are on Mac OSX or if all else fails just download the zip from the Apache website.

So first things first... why should you be excited about his?

• If you're already using Cassandra your data is already distributed and replicated, the Cassandra connector for Spark is aware of this distribution and can bring the computation to the data, this means it is going to be FAST
• Scala and the JVM might seem scary at first but Scala is an awesome language for writing data transformations
• The Spark-Shell: this is a REPL we can use for testing out code, simply put: it is awesome
• Spark can also connect with other data sources: files, RDMSs etc, which means you can do analytics combining data in Cassandra and systems like MySQL
• Spark also supports streaming, meaning we can combine new data in semi-real time with out batch processing
• Most importantly, you don't need to extract-transform-load your data from your operational database and put it in your batch processing system e.g. Hadoop

Lets get going

So what do we need to get all this magic working?

• Java - Java 7 or 8
• SBT - any 0.13.* will work. This is build tool used by the majority of Scala projects (Spark be Scala)
• Scala - Spark doesn't officially support 2.11 yet so get 2.10
• A Cassandra cluster
• A Spark installation (we're going simple this time so all on one computer)
• The Cassandra Spark connector with all of its dependencies bundled on the the classpath of the spark-shell for interactive use
• A fat jar with all our dependencies if we want to submit a job to a cluster (for the next post)

Hold up: jargon alert. Bundled dependencies, classpath? Fatman?

Both Cassandra and Spark run on the JVM, we don't really care about Cassandra and we're not submitting code to run inside Cassandra, but that is exactly what we're going to do with Spark.

That means all the code and libraries that we use are going to have to go everywhere our computation goes. This is because Spark distributes your computation across a cluster of computers. So we have to be kind and bundle all our code + all the dependencies we use (other jar files e.g for logging). The JVM classpath is just how you tell the JVM where all your jars are. 

Getting the Spark-Cassandra connector on the classpath

If you're from JVM land you probably are used to doing things like "just build a fat jar and put it on the classpath" if you're not then that is just a lot of funny words. So the connector is not part of core Spark, so you can't use it by default in the spark-shell. To do that you need to put the connector and all its dependencies on the classpath for the spark-shell. This sounds tedious right? You'd have to go and look at the build system of the connector and work out what it depends on. Welcome to JVM dependency hell. 

SBT, Maven, Gradle to the rescue (sort of). Virtually all JVM languages have a build system that allow you to declare dependencies, then it is the build system's responsibility to go get them from magic online locations (maven central) when you build your project. In Scala land this is SBT + Ivy.

When you come to distribute a JVM based application it is very kind to your users to build a far jat, or an "executable jar". This contains your code + all your dependencies so that it runs by its self, well apart from depending on a Java Runtime. 

So what we need to do is take the connector and use SBT + the assembly plugin to build our selves a fat jar. The Spark-Cassandra connector already has all the necesary config in its build scripts so we're just going to check it out and run "sbt assembly".

Lets take this line by line:

1. Line 1: Clone the Spark-Connector repo
2. Line 11: Run the SBT assembly command
3. Wait for ages
4. Line 14: Tells us where SBT has put the fat jar

Now it is time to use this jar in the Spark Shell:

Nothing fancy here, just gone into the bin directory of where I unzipped Spark and ran spark-shell --help. The option we're looking for is --jars. This is how we add our magical fat jar onto the classpath of the spark-shell. If we hadn't built a fat jar we'd be adding 10s of jars here!

However before we launch spark-shell we're going to add some properties to tell spark where Cassandra is, in the file (you'll need to create it): {Spark Install}/conf/spark-defaults.conf add:

 spark.cassandra.connection.host=192.168.10.11

Replace the IP with localhost if your Cassandra cluster is running locally. Then start up Spark-shell with the --jars option:

Now lets look at the important bits:

1. Line 1: Starting spark-shell with --jars pointing to the fat jar we built
2. Line 10: Spark confirming that it has picked up the connector fat jar
3. Line 11: Spark confirming that it has created us a SparkContext
4. Line 13: Import the connector classes, Scala has the ability to extend existing classes. The effect of this import is that we now have cassandra methods on our SparkConext
5. Line 16: Create a Spark RDD from a Cassandra table "kv" in the "test" keyspace
6. Line 19: Turn the RDD into an array (forcing it to complete the execution) and print the rows

Well that's all folks, next post will be about submitting jobs rather than using the spark-shell.

Akka: Testing messages sent to an actor's parent

I've blogged previously about testing messages sent back to the sender and to a child actor. The final common scenario I come across is testing actors that send a messages back to its supervisor/parent. If it is in response to a message from the parent you can use the same technique as described in testing messages sent back to the sender.

However, if that is not the case then TestKit has a very simple way you can do it with the TestActorRef and a TestProbe. Any where you have a piece of code that looks like this:

context.parent ! "Any message"

For example:

Then you can test it by passing a TestProbe to your TestActorRef and using the expect* functions on the TestProbe. So to test the actor above your test will look something like this:

As you can see the TestActorRef can take an ActorRef to use as a parent when you use the apply method that takes in a Props.

Full code here. Happy testing.

Testing Scala Futures with ScalaTest 2.0

Recently I was writing integration tests for a library that returned futures. I wanted simple synchronous tests.

Fortunately in the latest release of ScalaTest they have added something to do just that: the ScalaFutures trait

Say you have a class that returns a Future, e.g:

And you want to write a test for this and you want to wait for the future to complete. In regular code you might use a call back or a for comprehension e.g:


This won't work in a test as the test will finish before the future completes and the assertions are on a different thread.

ScalaFutures to the rescue! Upgrade to ScalaTest 2.0+ and mix in the ScalaFutures trait. Now you can test using whenReady:

Or with futureValue:

Happy testing! Full source code here.

Akka: Testing messages sent to child actors

I previously blogged about testing messages sent back to the sender. Another common scenario is testing that an actor sends the correct message to a child. A piece of code that does this might look like:

But how do you test drive this kind of code?

I've adopted the approach of removing the responsibility of creating the child from the ParentActor. To do this you can pass in a factory to the ParentActor's constructor that can be replaced for testing. To do this the ParentActor is changed to look like this:

The ParentActor now takes in a factory function which takes in an ActorFactoryRef and returns a ActorRef. This function is responsible for creating the ChildActor.

childFactory: (ActorRefFactory) => ActorRef

This is then used by the ParentActor as follows:

val childActor = childFactory(context)

You can now inject a TestProbe for testing rather than using a real ActorFactorRef. For example:

And for the real code you can pass in an anonymous function that uses the ActorRefFactory to actually create a real ChildActor:

Here you can see the ActorFactoryRef (context in Actors implement this interface) is used to create an Actor the same way as the original ParentActor did.

Full source code here.

Akka: Testing that an actor sends a message back to the sender

A common pattern when using actors is for actor A to send a message to Actor B and then for Actor B to send a message back.

This can be tested using the Akka testkit along with Scala Test.

Take the following very simple example. We have an Actor called Server which should accept Startup messages and respond with a Ready message when it is ready for clients to send additional messages.

We can start with a noddy implementation that does nothing:

Then write a test to specify the behaviour. Here I've used TestKit along with Scala Test FunSuite.

This test will fail with the following error message:

assertion failed: timeout (3 seconds) during expectMsg while waiting for Ready

As you can probably guess TestKit waited for 3 seconds for a Ready message to be sent back.

To fix the test we add the the following to the Server Actor implementation:

And now the test will pass! The two important things to take note of are that our test case extended from TestKit, this gives you an ActorSystem. And that the test case mixed in the ImplicitSender trait, this allows us to receive messages use the methods like "expectMsg" to assert that the correct message has been received.

Scala, MongoDB and Casbah: Dealing with Arrays

Get hold of a collection object using something like this:

scala> val collection = MongoClient()("test")("messages")

collection: com.mongodb.casbah.MongoCollection = messages

Where test is the database and messages is the name of the collection.

Inserting arrays is nice and easy, just build up your MongoDBObject with Lists inside:

scala> collection.insert(MongoDBObject("message" -> "Hello World") ++ ("countries" -> List("England","France","Spain")))

res18: com.mongodb.casbah.Imports.WriteResult = { "serverUsed" : "/127.0.0.1:27017" , "n" : 0 , "connectionId" : 234 , "err" :  null  , "ok" : 1.0}

Use your favourite one liner to print all the objects in the collection:

scala> collection.foreach(println(_))

{ "_id" : { "$oid" : "523f145e30041dae32fd04da"} , "message" : "Hello World" , "countries" : [ "England" , "France" , "Spain"]}

Now lets say you want a list of objects, simply create a list of MongoDBObjects:

scala> collection.insert(MongoDBObject("message" -> "A list of objects?") ++ ("objects" -> List(MongoDBObject("name" -> "England"),MongoDBObject("name" -> "France"))))

res20: com.mongodb.casbah.Imports.WriteResult = { "serverUsed" : "/127.0.0.1:27017" , "n" : 0 , "connectionId" : 234 , "err" :  null  , "ok" : 1.0}

scala> collection.foreach(println(_))

{ "_id" : { "$oid" : "523f145e30041dae32fd04da"} , "message" : "Hello World" , "countries" : [ "England" , "France" , "Spain"]}

{ "_id" : { "$oid" : "523f14b530041dae32fd04db"} , "message" : "A list of objects?" , "objects" : [ { "name" : "England"} , { "name" : "France"}]}

Now reading them back out of Mongo and processing the array items individually. First we can get a hold of an object that contains an array:

scala> val anObjectThatContainsAnArrayOfObjects = collection.findOne().get

anObjectThatContainsAnArrayOfObjects: collection.T = { "_id" : { "$oid" : "523f145e30041dae32fd04da"} , "message" : "Hello World" , "countries" : [ "England" , "France" , "Spain"]}

The extra get is on the end as we used the findOne method this time and it returns an Option. Then we can get just the array field:

val mongoListOfObjects = anObjectThatContainsAnArrayOfObjects.getAs[MongoDBList]("countries").get

mongoListOfObjects: Option[com.mongodb.casbah.Imports.MongoDBList] = Some([ "England" , "France" , "Spain"])

Now we have a handle on a MongoDBList which represents our array in Mongo. The MongoDBList is Iterable so we can loop through and print it out:

scala> mongoListOfObjects.foreach( country => println(country) )

England

France

Spain

Or map it to a sequence of Strings:

scala> val listOfCountries = mongoListOfObjects.map(_.toString)

listOfCountries: scala.collection.mutable.Seq[String] = ArrayBuffer(England, France, Spain)

scala> listOfCountries

res24: scala.collection.mutable.Seq[String] = ArrayBuffer(England, France, Spain)

Scala and MongoDB: Getting started with Casbah

The officially supported Scala driver for Mongo is Casbah. Cashbah is a thin wrapper around the Java MongoDB driver that gives it a Scala like feel. As long as you ignore all the MongoDBObjects then it feels much more like being in the Mongo shell or working in Python that working with Java/Mongo.

All the examples are copied from a Scala REPL launched from an SBT project with Casbah added as a dependency.

So lets get started by importing the Casbah package:

scala> import com.mongodb.casbah.Imports._ 
import com.mongodb.casbah.Imports._

Now lets create a connection to a locally running Mongo and use the "test" database:

scala> val mongoClient = MongoClient()

mongoClient: com.mongodb.casbah.MongoClient = com.mongodb.casbah.MongoClient@2acf0276

scala> val database = mongoClient("test")

database: com.mongodb.casbah.MongoDB = test

And now lets get a reference to the messages collections:

scala> val collection = database("messages")

collection: com.mongodb.casbah.MongoCollection = messages

As you can see Casbah makes heavy use of the apply method to give relatively nice boiler plate connection code. To print all the rows for a collection you can use the find method which returns an iterator (there is none at the moment):

scala> collection.find().foreach(row => println(row) )

Now lets insert some data the using the insert method and then find and print it:

scala> collection.insert(MongoDBObject("message" -> "Hello world"))

res2: com.mongodb.casbah.Imports.WriteResult = { "serverUsed" : "/127.0.0.1:27017" , "n" : 0 , "connectionId" : 225 , "err" :  null  , "ok" : 1.0}

scala> collection.find().foreach(row => println(row) )

{ "_id" : { "$oid" : "523aa69a30048ee48f49c333"} , "message" : "Hello world"}

And adding another document:

scala> collection.insert(MongoDBObject("message" -> "Hello London"))

res4: com.mongodb.casbah.Imports.WriteResult = { "serverUsed" : "/127.0.0.1:27017" , "n" : 0 , "connectionId" : 225 , "err" :  null  , "ok" : 1.0}

scala> collection.find().foreach(row => println(row) )

{ "_id" : { "$oid" : "523aa69a30048ee48f49c333"} , "message" : "Hello world"}

{ "_id" : { "$oid" : "523aa6bf30048ee48f49c334"} , "message" : "Hello London"}

The familiar findone method is there. Rather than an Iterable object returned from find, findOne returns an Option so you can use a basic pattern match to handle the document being there or not:

scala> val singleResult = collection.findOne()

singleResult: Option[collection.T] = Some({ "_id" : { "$oid" : "523aa69a30048ee48f49c333"} , "message" : "Hello world"})

scala> singleResult match {

     |   case None => println("No messages found")

     |   case Some(message) => println(message)

     | }

{ "_id" : { "$oid" : "523aa69a30048ee48f49c333"} , "message" : "Hello world"}

Now lets query using the ID of an object we've inserted (querying by any other field is the same):

scala> val query = MongoDBObject("_id" -> helloWorld.get("_id"))

id: com.mongodb.casbah.commons.Imports.DBObject = { "_id" : { "$oid" : "523aa69a30048ee48f49c333"}}

scala> collection.findOne(query)

res12: Option[collection.T] = Some({ "_id" : { "$oid" : "523aa69a30048ee48f49c333"} , "message" : "Hello world"})

We can also update the document in the database and then get it again to prove it has changed:

scala> collection.update(query, MongoDBObject("message" -> "Hello Planet"))

res13: com.mongodb.WriteResult = { "serverUsed" : "/127.0.0.1:27017" , "updatedExisting" : true , "n" : 1 , "connectionId" : 225 , "err" :  null  , "ok" : 1.0}

scala> collection.findOne(query)

res14: Option[collection.T] = Some({ "_id" : { "$oid" : "523aa69a30048ee48f49c333"} , "message" : "Hello Planet"})

The remove method works in the same way, just pass in a MongoDBObject for the selection criterion.

Not look Scalary enough for you? You can also insert using the += method:

scala> collection += MongoDBObject("message"->"Hello England")

res15: com.mongodb.WriteResult = { "serverUsed" : "/127.0.0.1:27017" , "n" : 0 , "connectionId" : 225 , "err" :  null  , "ok" : 1.0}

scala> collection.find().foreach(row => println(row))

{ "_id" : { "$oid" : "523aa69a30048ee48f49c333"} , "message" : "Hello Planet"}

{ "_id" : { "$oid" : "523aa6bf30048ee48f49c334"} , "message" : "Hello London"}

{ "_id" : { "$oid" : "523c911230048ee48f49c335"} , "message" : "Hello England"}

How do you build more complex document in Scala? Simply use the MongoDBObject ++ method, for example we can create an object with multiple fields, insert it, then view it by printing all the documents in the collection:

scala> val moreThanOneField = MongoDBObject("message" -> "I'm coming") ++ ("time" -> "today") ++ ("Name" -> "Chris")

moreThanOneField: com.mongodb.casbah.commons.Imports.DBObject = { "message" : "I'm coming" , "time" : "today" , "Name" : "Chris"}

scala> collection.insert(moreThanOneField)

res6: com.mongodb.casbah.Imports.WriteResult = { "serverUsed" : "/127.0.0.1:27017" , "n" : 0 , "connectionId" : 234 , "err" :  null  , "ok" : 1.0}

scala> collection.find().foreach(println(_) )

{ "_id" : { "$oid" : "523aa69a30048ee48f49c333"} , "message" : "Hello Planet"}

{ "_id" : { "$oid" : "523aa6bf30048ee48f49c334"} , "message" : "Hello London"}

{ "_id" : { "$oid" : "523c911230048ee48f49c335"} , "message" : "Hello England"}

{ "_id" : { "$oid" : "523c96b530041dae32fd04d6"} , "message" : "I'm coming" , "time" : "today" , "Name" : "Chris"}

Scala: What logging library to use?

Having investigated a few options I have decided on SLF4J + Logback + Grizzled. The project I am currently on uses Scalatra - this matches their solution.

Three libraries?? That sounds overkill but it is in fact very simple.

SLF4J + Logback are common place in Java projects. SLF4J is logging facade - basically a set of interfaces to program to where as Logback is an implementation you put on your classpath at runtime. There are other implementations of SLF4J you can use such as Log4J. Grizzled is a Scala wrapper around SLF4J to give it Scala like usage.

Having worked on many Java projects that use SLF4J + Logback I'm used to seeing lines at the top of files that look like this:

private static final Logger LOGGER = LoggerFactory.getLogger(SomeClass.class)

Fortunately Grizzled-SLF4J + traits help here.

Mixing in the Grizzled Logging trait allows you to write logging code like this:

This will produce logs like this:

09:57:11.169 [main] INFO com.batey.examples.SomeClass - Some information
09:57:11.172 [main] ERROR com.batey.examples.SomeClass - Something terrible

The grizzled trait uses your class name as the logger name. Job done with less boilerplate than Java!

Everything you need s on Maven central so can be added to your pom or sbt dependencies:
Grizzled: Grizzled
Logback: Logback

Scala: SBT OutOfMemoryError: PermGen space

When starting out with Scala/SBT I very quickly ran into perm gen issues followed by SBT crashing:

java.lang.OutOfMemoryError: PermGen space

Especially when running the console from within interactive mode.

To fix/brush this under the carpet add the following to your profile (.bashrc / .bash_profile) and source it again (run . ~/.bashrc)

export SBT_OPTS=-XX:MaxPermSize=256m

7 languages in 7 weeks: Scala day 2

Day two is all about collections. Forcing the typical Java developer (me) to think about iteration in a different way.

Exercise 1 is to take a List of strings and add up the number of characters using the fold left function. There is a very similar example in the book so I think the author is just checking you are paying attention:

Exercise 2 introduces you to Traits by asking you to develop a trait that has a method which censors a string by substituting words in a string. The words to substitute and what to substitute them with should be stored in a Map.


Then create a version that loads the substitutions from a file:

I chose to extend the trait and populate the alternatives field. Spending all my days writing Java what I ended up producing looks very alien to me. It works (I think) something like this:

• Open and read a file using the library method fromFile
• Get the lines of the file one by one using getLines
• Map each like to an Array by splitting the like with the = sign (the file is key value pairs like a Java properties file)
• Map the array to a tuple then turn it into a list
• Construct a map from the List of tuples

And done! Looks a lot nicer than it sounds! The full source code is available on my github.

7 languages in 7 weeks: Scala day 1

I went to a great talk by Sandro Mancuso at Devoxx UK last week. He mentioned the project he is currently working on being written in Java for the main business flow and Scala for the data manipulation (sorry if I have remembered that incorrectly). This got me thinking it was about time I learnt some Scala. I've dipped in and out of the 7 languages in 7 weeks book; fortunately it has a chapter on Scala. So here I go...

 Googling around and reading the first day the following stands out:

• Type inference with the safety of static typing (we're all sick of typing types aren't we??)
• Companion objects instead of static methods
• Higher order functions (why do we need lambas? Can't we just switch to Scala?)
• Interfaces, no I mean mixins, wait traits: interfaces with implementation? I think these are great when used well.
• Tuples and ranges.

So on to my home work!

The exercise for day 1 is to write noughts and crosses. I decided to concentrate on setting up a nice environment for writing scala rather than great code. I started with vim, then sublime but decided I fancied using an IDE. I started with Eclipse but moved to IntelliJ as it looks nicer on a retina screen :) In  addition to setting up an IDE I wanted to do TDD from the start, so I got to grips with ScalaTest.

Here are a couple of example tests:


Here is the main Board object. I could pull out the logic to determine a winner into a class like Game but as this was my first Scala code I decided to keep it simple in a single file.


I've omitted the code that calculates the different types thought it's all available on github. I looked at using a case class rather than a class that extends Enumeration but I haven't been taught case classes yet  so I'll leave that until later.

That's it for day one!