Skip to content


Subversion checkout URL

You can clone with
Download ZIP
Storm Cassandra Integration

Fetching latest commit…

Cannot retrieve the latest commit at this time

Failed to load latest commit information.

Storm Cassandra Integration

Integrates Storm and Cassandra by providing a generic and configurable backtype.storm.Bolt implementation that writes Storm Tuple objects to a Cassandra Column Family.

How the Storm Tuple data is written to Cassandra is dynamically configurable -- you provide classes that "determine" a column family, row key, and column name/values, and the bolt will write the data to a Cassandra cluster.

Project Location

Primary development of storm-cassandra will take place at:

Point/stable (non-SNAPSHOT) release souce code will be pushed to:

Maven artifacts for releases will be available on maven central.

Building from Source

    $ mvn install


Basic Usage

CassandraBolt expects that a Cassandra hostname, port, and keyspace be set in the Storm topology configuration:

    Config config = new Config();
    config.put(CassandraBolt.CASSANDRA_HOST, "localhost");
    config.put(CassandraBolt.CASSANDRA_PORT, 9160);
    config.put(CassandraBolt.CASSANDRA_KEYSPACE, "testKeyspace");

The CassandraBolt class provides a convenience constructor that takes a column family name, and row key field value as arguments:

    IRichBolt cassandraBolt = new CassandraBolt("columnFamily", "rowKey");

The above constructor will create a CassandraBolt that writes to the "columnFamily" column family, and will look for/use a field named "rowKey" in the backtype.storm.tuple.Tuple objects it receives as the Cassandra row key.

For each field in the backtype.storm.Tuple received, the CassandraBolt will write a column name/value pair.

For example, given the constructor listed above, a tuple value of:

    {rowKey: 12345, field1: "foo", field2: "bar}

Would yield the following Cassandra row (as seen from cassandra-cli):

    RowKey: 12345
    => (column=field1, value=foo, timestamp=1321938505071001)
    => (column=field2, value=bar, timestamp=1321938505072000)


The "examples" directory contains two examples:

  • CassandraReachTopology

  • PersistentWordCount

Cassandra Reach Topology

The CassandraReachTopology example is a Storm Distributed RPC example that is essentially a clone of Nathan Marz' ReachTopology, that instead of using in-memory data stores is backed by a Cassandra database and uses generic storm-cassandra bolts to query the database.

Persistent Word Count

The sample PersistentWordCount topology illustrates the basic usage of the Cassandra Bolt implementation. It reuses the TestWordSpout spout and TestWordCounter bolt from the Storm tutorial, and adds an instance of CassandraBolt to persist the results.


In order to run the examples, you will need a Cassandra database running on localhost:9160.

Build the Example Source

    $ cd examples
    $ mvn install

Create Cassandra Schema and Sample Data

Install and run Apache Cassandra.

Create the sample schema using cassandra-cli:

    $ cd schema
    $ cassandra-cli -h localhost -f cassandra_schema.txt

Running the Cassandra Reach Topology

To run the CassandraReachTopology execute the following maven command:

    $ mvn exec:java -Dexec.mainClass=backtype.storm.contrib.cassandra.example.CassandraReachTopology

Among the output, you should see the following:

Reach of 3
Reach of 3
Reach of 4
Reach of 0

To enable logging of all tuples sent within the topology, run the following command:

    $ mvn exec:java -Dexec.mainClass=backtype.storm.contrib.cassandra.example.CassandraReachTopology -Ddebug=true

Running the Persistent Word Count Example

The PersistentWordCount example build the following topology:

TestWordSpout ==> TestWordCounter ==> CassandraBolt

Data Flow

  1. TestWordSpout emits words at random from a pre-defined list.
  2. TestWordCounter receives a word, updates a counter for that word, and emits a tuple containing the word and corresponding count ("word", "count").
  3. The CassandraBolt receives the ("word", "count") tuple and writes it to the Cassandra database using the word as the row key.

Run the PersistentWordCount topology:

    $ mvn exec:java -Dexec.mainClass=backtype.storm.contrib.cassandra.example.PersistentWordCount

View the end result in cassandra-cli:

    $ cassandra-cli -h localhost
    [default@unknown] use stormks;
    [default@stromks] list stormcf;

The output should resemble the following:

    Using default limit of 100
    RowKey: nathan
    => (column=count, value=22, timestamp=1322332601951001)
    => (column=word, value=nathan, timestamp=1322332601951000)
    RowKey: mike
    => (column=count, value=11, timestamp=1322332600330001)
    => (column=word, value=mike, timestamp=1322332600330000)
    RowKey: jackson
    => (column=count, value=17, timestamp=1322332600633001)
    => (column=word, value=jackson, timestamp=1322332600633000)
    RowKey: golda
    => (column=count, value=31, timestamp=1322332602155001)
    => (column=word, value=golda, timestamp=1322332602155000)
    RowKey: bertels
    => (column=count, value=16, timestamp=1322332602257000)
    => (column=word, value=bertels, timestamp=1322332602255000)

    5 Rows Returned.
    Elapsed time: 8 msec(s).
Something went wrong with that request. Please try again.