Installing Cassandra and Spark with Ansible

I am currently doing a proof of concept with Spark and Cassandra. I quickly need to be able to create and start Cassandra and Spark clusters. Ansible to the rescue!

I split this my ansible playbook into three roles:

• Cassandra
• Ops center
• Spark

My main playbook is very simple:

I have some hosts defined in a separate hosts file called m25-cassandra. I've decided to install htop, I could have out this in a general server role.

I also define a few variables, these of course course could be defined else where per role:

• cluster_name - this will replace the cluser name in each of the hosts cassandra.yaml
• seeds - as above

So lets take a look at each role.

Cassandra

Here are the tasks:

This is doing the following:

• Installing a JRE
• Adding the Apache Cassandra debian repository
• Adding the keys for the debian repository
• Installing the latest version of Cassandra
• Replacing the cassandra.yaml (details later)
• Ensuring Cassandra is started

The template cassandra.yaml uses the following variables:

• cluster_name: '{{ cluster_name }}' - So we can rename the cluster
• - seeds: "{{ seeds }}" - So when we add a new node it connects to the cluster
• listen_address: {{ inventory_hostname }} - Listen on the nodes external IP so other nodes can communicate with it
• rpc_address: {{ inventory_hostname }} - So we can connect ops center and cqlsh to the nodes

Magic! Now adding new hosts to my hosts file with the tag m25_cassandra will get Cassandra installed, connected to the cluster and started.

Ops Center

The tasks file for ops center:

This is doing the following:

• Adding the Datastax community debian repository
• Adding the key for the repo
• Installing Ops Center
• Starting Ops Center

No templates here as all the default configuration is fine.

Spark

The spark maven build can build a debian package but I didn't find a public debian repo with it in so the following just downloads and unzips the Spark package:

I start the workers using the start-slaves.sh script from my local master do don't need to start anything on the nodes that have Cassandra on.

Conclusion

Ansible makes it very easy to install distributed systems like Cassandra. The thought of doing it manually fills me with pain! This is just got a PoC, I don't suggest downloading Spark from the public internet or always installing the latest version of Cassandra for your production systems. The full souce including templates and directory structure is here.

Cassandra: Datastax Java driver retry policy

The Datastax Java Driver for Cassandra exposes its strategy for retrying via the following interface:

There are three scenarios you can control retry policy for:

1. Read time out: When a coordinator received the request and sent the read to replica(s) but the replica(s) did not respond in time
2. Write timeout: As above but for writes
3. Unavailable: When the coordinator is aware there aren't enough replica available without sending the read/write request on

What is the default behaviour?

The DefaultRetryPolicy retries with the following behaviour:

1. Read timeout: When enough replica are available but the data did not come back within the configured read time out 
2. Write timeout: Only if the initial phase of a batch write times out - see cassandra batch statement
3. Unavailable timeout: Never

How do I configure the value for the read and write timeout?

This is configured in the cassandra.yaml on the Cassandra server.  The default is 10 seconds, you can change the following properties:

# How long the coordinator should wait for read operations to complete
read_request_timeout_in_ms: 10000
# How long the coordinator should wait for writes to complete
write_request_timeout_in_ms: 10000

What are the other policies?

DowngradingConsistencyRetryPolicy

The most complicated retry policy and comes with a big warning: your read/write may be re-tried at a lower consistency. So if you have business requirements to not report success if you don't meet a certain level of consistency then use this with cation.

What does it do?

1. Read: If at least one replica responded then the read is retried at a lower consistency
2. Write: Retries for unlogged batch queries when at least one replica responded (see http://www.datastax.com/dev/blog/atomic-batches-in-cassandra-1-2) and for all other types of writes the timeout is just ignored if at least one replica acknowledged the write (essentially ignoring the consistency request)
3. Unavailable: If at least one replica is available then the query is re-tried with a lower consistency

FallthroughRetryPolicy

No retrying! Any failure is re-thrown to the client.

LoggingRetryPolicy

This is just a decorator policy that you can wrap around any other policy that logs ignored (no retry) and any actual retries. The driver uses SLF4J and logs at INFO level.

How do I use a different policy?

Simply add it with creating your Cluster. The retry policies all have a singleton you can use e.g:

Conclusion

The Datastax driver is very open for extension as it exposes its strategies for retry, load balancing and reconnection. 

The retry policy is very easy to work with as all the current implementations are stateless. I'll follow this post up with how to implement your own retry policy.

Talk: Introduction to Cassandra, CQL, and the Datstax Java Driver

Presenter: Johnny Miller

Who is he? Datastax Solutions Architect

Where? Skills matter exchange in London

I went to an "introductory" talk even though I have a lot of experience with Cassandra for a few reasons:

• Meet other people in London that are using Cassandra
• To discover what I don't know about Cassandra

Here are my notes that in roughly the same order as the talk.

What's Cassandra? The headlines

• Been around for ~5 years - originally developed by Facebook for their inbox search
• Distributed key store - column orientated data model
• Tuneable consistency - per request decide how consistent you want the response to be
• Datacenter aware with asynchronous replication
• Designed for use as a cluster - not much value in a single node Cassandra deployment

Gossip - how nodes in a cluster learn about other nodes

• P2P protocol for how nodes discover location and state of other nodes
• New nodes are given seed nodes for bootstrapping - but these aren't single points of failure as they aren't used again

Data distribution and replication

• Replication factor: How many nodes each piece of data is stored on
• Each node is given a range of primary keys to look after

Partitioners - How to decide which node gets what data

• Row keys are hashed to decide node then a replication strategy defines how to pick the other replicas

Replicas - how to select where else the data lives

• All replicas are equally important. No difference between the node the key hashed to and the other replicas that were selected
• Two ways to pick the other replicas:
• Simple: Only single DC. Specify just a replication factor. Hashes the key and then walks the cluster and picks the replicas. Not very clever - all replicas could end up on the same rack
• Network: Configure with a RF per DC. Walk the ring for each DC until it reaches a node in another rack

Snitch - how to define a data centre and a rack

• Informs Cassandra about node topology, designates DC and Rack for every node
• Example: Rack inferring snitch designates DC and Rack based on the IP of the node 
• Example: Property file snitch where every node has the DC and Rack of every other node
• Example: GossipingPropertyFileSnitch: Every node knows its own DC and Rack and tells other nodes via Gossip
• Dynamic snitching: monitors performance of reads, this snitch wraps the other snitches to respond to network latency

Client requests

• Connect to any client in the node - becomes the coordinator. This node knows which nodes to talk to for the request
• Multi DC - picks a coordinator in the other data centre  to replicate data there or to get data for a read

Consistency

• Quorum = (Replication Factor/2) + 1 i.e. more than half
• E.g R = 3, Q = 2, tolerate 1 replica going down to continue reading and writing at Quorum
• Per request consistency - can decide certain writes are more important and require higher consistency than others
• Example consistency levels: ANY, ONE, TWO, THREE, QUORUM, EACH_QUORUM, LOCAL_QUORUM
• SERIAL: New in cassandra 2.0

Write requests - what happens?

• The coordinator (node the client connects to) forwards the write to all the replicas in the local DC and designates a coordinator in the other DCs to do the same there
• The coordinator may be a replica but does not need to be
• For a single node writes first go to commit log (disk), then writes to meltable (memory)
• When does the write succeed? Depends on consistency e.g a write consistency of ONE means that the data needs to be in the commit log and memtable of at least one replica

Hinted handoff - how Cassandra deals with nodes being down on write

• Coordinator node keeps hints if one of the replicas down
• When the node comes back up the hints are then sent to the node so it can catch up
• Hints are kept for a finite amount of time - default is three hours

Read requests  - what happens?

• Coordinator contacts a number of nodes depending on the consistency - once enough have responded the read can be successful 
• Will send requests to node responding the fastest
• If not consistent - compare timestamps + do a read repair
• Possible other background read repair

What was missing?

Overall it was a great talk however here is some possible improvements:

• A glossary/overview at the start? Perhaps a mapping from relational terminology to Cassandra terminology. For example the term keyspace was used a number of times before describing what it is
• Overview of consistency when talking about eventual consistency - however this did come later? A few scenarios for when read/writes at different consistency levels would fail/succeed would have been very helpful 
• Compaction required for an intro talk? I thought talking about compaction was a bit too much for an introductory talk as you need to understand memtables and sstables before it makes sense
• The downsides of Cassandra: for example some forms of schema migration/change is a nightmare when you are using CQL3 + have data you need to migrate

Time series based queries in Cassandra 1.2+ and CQL3

Cassandra is fantastic for storing large amounts of data, and as of 1.2/CQL3 working with time series data just got a lot easier. Here is a basic example that stores some kind of posts (e.g blog) that can be queried by username and a time period.

Everything is assuming you have Cassandra 1.2+ installed and are using CQL3 with the cqlsh python client. Here are the exact versions I used for the below example:

[cqlsh 3.0.2 | Cassandra 1.2.5 | CQL spec 3.0.0 | Thrift protocol 19.36.0]

Creating a keyspace:
create keyspace cassandraspike WITH REPLICATION = {'class' : 'SimpleStrategy', 'replication_factor': 1};

The syntax has changed for creating keyspaces in CQL for cassandra 1.2 so if this fails check you aren't running against a pre-1.2 version of cassandra.
use cassandraspike;

Then create a posts table:
create table posts(username varchar, time timeuuid, post_text varchar, primary key(username, time))

We've used a compound primary key that is the username and the time. The time is of type timeuuid which is a new data type in CQL3 that us a type 1 UUID that contains a timestamp so it both stores the time of the post and makes our rows unique.

If you're familier with column families then it might interest you to know that the first column in your primary key becomes the column family (CF) row key. Every subsequent part of the primary key becomes part of the CF column names in that CF row. This has two implications:

• There will only be as many CF rows as there are variations of the first element in your primary key. This can be a problem if this element has a very low cardinality as you can end up with very wide CF rows.
• CF columns are stored in order so the fact that each CF column is prefixed with the time means that your data is stored in the order it happened. Making it very efficient to be queried by time.

Hopefully you are sufficiently convinced that your data is going to be stored in a way that is efficient to query by time. So let's store some data in and query it by user name and time.

To insert we'll make use if the now() function which gives you a timeuuid for the current time:

insert into posts(username, time, post_text) values ('chbatey', now(), 'i am writing something about cassandra');

Selecting this back works but gives you a time which isn't too human read able:
select * from posts;

 username | time                                 | post_text
----------+--------------------------------------+----------------------------------------
  chbatey | 59ad61d0-c540-11e2-881e-b9e6057626c4 | i am writing something about cassandra


That's where the dateOf function comes in handy which converts a timeuuid into a date:
select username, dateOf(time), post_text from posts;

 username | dateOf(time)             | post_text
----------+--------------------------+----------------------------------------
  chbatey | 2013-05-25 14:38:14+0100 | i am writing something about cassandra

I've now inserted another post at a different time:
select username, dateOf(time), post_text from posts;


username | dateOf(time)             | post_text
----------+--------------------------+----------------------------------------
  chbatey | 2013-05-25 14:38:14+0100 | i am writing something about cassandra
  chbatey | 2013-05-25 14:40:35+0100 | i am writing something about cassandra


Now lets say you are only interested in the posts chbatey made at 14:38:
select username, dateOf(time), post_text from posts where time >= minTimeuuid('2013-05-25 14:38') and time < minTimeuuid('2013-05-25 14:39') and username = 'chbatey';

 username | dateOf(time)             | post_text
----------+--------------------------+----------------------------------------
  chbatey | 2013-05-25 14:38:14+0100 | i am writing something about cassandra

This query is more complicated. It makes use of another cql function: minTimeuuid. The minTimeuuid function gives a fake (as it isn't unique) timeuuid that is the smallest possible timeuuid for the given time. This is very hand when you want to do less than/greater than queries on timeuuid fields.

In the above query we are getting everything that is greater than the minimum timeuuid for the time 2013-05-25 14:38 but less than the minimum timeuuid of the next minute 2013-05-25 14:39.

Rather than that we could have used the very similar function maxTimeuuid function with 2013-05-24 14:39:59. However I find original one easier to understand as I read it as greater than or equal to 14:38:00 and less than 14:39:00.