Persistence

Persistent datastructure API

Akka provides a set of persistent datastructures:

• PersistentMap (implements 'scala.collection.mutable.Map')
• PersistentVector (implements 'scala.RandomAccessSeq')
• PersistentRef (CAS cell)

All these are fully managed by the STM (see the STM section for details) which means that you have to be the context of a transaction in order to read or write to these datastructures. If you try to access one of them outside a transaction, an exception will be thrown. The PersistentMap and PersistentVector gives a feel of being mutable datastructures but the are using immutable data (the persistent datastructures HashTrie and Vector) and Managed References under the hood. PersistentRef however must always be used with only immutable data.

The persistence module is in itself modular and has support for many pluggable backend storages. In order to create and retrieve a persistent datastructure you have to use the 'xStorage' factory object for the storage backend you are interested in.

You can create and retrieve a Map, Vector and Ref based on an explicit user-defined id. This is best since it gives you best control and an ability to lookup the datastructure when you need it, even on another node. But if you don't specify an explicit id then a UUID will be created for you.

All these examples are using the 'CassandraStorage' factory object but work equally the same with any supported backend.

Create a new persistent datastructure based on an explicit id:

val map =    CassandraStorage.newMap(id)
val vector = CassandraStorage.newVector(id)
val ref =    CassandraStorage.newRef(id)

Retrieve a persistent datastructure based on an explicit id (will be created if it doesn't exist):

val map =    CassandraStorage.getMap(id)
val vector = CassandraStorage.getVector(id)
val ref =    CassandraStorage.getRef(id)

Create a new persistent datastructure based on a generated random id:

val map =    CassandraStorage.newMap
val vector = CassandraStorage.newVector
val ref =    CassandraStorage.newRef

Cassandra

Akka supports the excellent Cassandra distributed structured storage database as a pluggable backend alternative. It is a hybrid between Google’s BigTable and Amazon’s Dynamo. It has BigTable’s data model so it is a hybrid between a column and a row database, giving the best of both worlds.

Create Cassandra based persistent datastructures

val map =    CassandraStorage.newMap(id)
val vector = CassandraStorage.newVector(id)
val ref =    CassandraStorage.newRef(id)

Akka’s Cassandra configuration

Akka’s Cassandra configuration resides in the ‘akka-reference.conf’ configuration file, and looks as follows:

 <storage>
    <cassandra>
      hostname = "127.0.0.1"        # IP address or hostname of one of the Cassandra cluster's seeds
      port = 9160
      storage-format = "scala-json" # Options: java, scala-json, java-json, protobuf
      consistency-level = "QUORUM"  # Options: ZERO, ONE, QUORUM, ALL
    </cassandra>
  </storage>

Most of it is self-explanatory except from the consistency-level. Which is described as follows:

• ZERO: Ensure nothing. A write happens asynchronously in background
• ONE: Ensure that the write has been written to at least 1 node’s commit log and memory table before responding to the client.
• QUORUM: Ensure that the write has been written to N/ 2 + 1 nodes before responding to the client.
• ALL: Ensure that the write is written to nodes before responding to the client.

Cassandra’s own configuration file

Cassandra’s configuration file resides in ‘$AKKA_HOME/config/storage-conf.xml’. This configuration file you have to move to the Cassandra distribution’s ‘./conf’ directory. Some of the things here are Akka specific and should not be touched but there are some option you have to edit if you for example want to run more than one node (which is really the point of Cassandra as well as Akka).
Akka is hiding the regular Cassandra API since it wraps it in transactional Map, Vector and Ref. However there are some things you should be aware of. The most important configuration setting that you have to change if you are running multiple nodes is:

<!--
     Addresses of hosts that are deemed contact points. Cassandra nodes use
     this list of hosts to find each other and learn the topology of the ring.
     You must change this if you are running multiple nodes!
-->
<Seeds>
    <Seed>127.0.0.1</Seed>
</Seeds>

Another thing you should change if you are running with more than a single node is the replication-factor. This setting defines how many replicas of the data that Cassandra should have. 1 means no replicas. If you are running in production then you should set it to 3.

<ReplicationFactor>3</ReplicationFactor>

Another thing you might want to edit is the different storage directories. It is usually best to put these on separate disks and for best performance you should make sure that the commit log directory and data directory are on different disks.

However, you should remove anything from the ‘Keyspaces’ settings, however you are free to add your own keyspace and column families for usage outside of Akka’s current transactional API.

<Keyspaces>
  <Keyspace Name="akka">
    <KeysCachedFraction>0.01</KeysCachedFraction>
    <ColumnFamily CompareWith="UTF8Type" Name="map" />
    <ColumnFamily CompareWith="UTF8Type" Name="vector"/>
    <ColumnFamily CompareWith="UTF8Type" Name="ref"/>
  </Keyspace>
</Keyspaces>

Here is an excellent article to read to get more familiar with Cassandra.

Cassandra Session API

Akka also provides a direct Cassandra API through a Scala as well as Java style session interface. This session abstracts away session pooling, protocol etc. and is a nice option if you need to by-pass the STM persistence abstractions (Map, Vector and Ref). The class you need to use is 'se.scalablesolutions.akka.state.CassandraSessionPool'.

Here is an example of usage:

// create the session pool
val sessions = new CassandraSessionPool(
  keyspace,
  StackPool(SocketProvider(hostname, port)),
  protocol,
  ConsistencyLevel.QUORUM)
 
// insert a column
sessions.withSession { session =>
  session ++| (key, new ColumnPath(columnFamily, null, columnName), serializer.out(element), System.currentTimeMillis)
}
 
// retrieve a column
val column: Option[ColumnOrSuperColumn] = sessions.withSession { session =>
  session | (key, new ColumnPath(columnFamily, null, columnName))
}
 
// get a range of colums
val columns: List[ColumnOrSuperColumn] = sessions.withSession { session =>
  session / (key, columnParent, slicePredicate, isAscending, count)
}

MongoDB

Akka also supports MongoDB as the persistent store for transactors. MongoDB is a high performance schema-free, document oriented data store that offers collection oriented storage and SQL like query facilities.

Create MongoDB based persistent datastructures

val map =    MongoStorage.newMap(id)
val vector = MongoStorage.newVector(id)
val ref =    MongoStorage.newRef(id)

MongoDB configuration

Akka’s MongoDB configuration is included as part of $AKKA_HOME/config/akka.conf. Please refer to the following section in the configuration file:

<storage>
  <mongodb>
    service = on
    hostname = "127.0.0.1"             # IP address or hostname of the MongoDB DB instance
    port = 27017                              # default port
    dbname = "mydb"                      # database name
    storage-format = "sjson"            # Options: java, scala-json, java-json, protobuf, sjson
  </mongodb>
</storage>

Running MongoDB server

MongoDB server has to be run in order to use MongoDB as the persistent storage for Akka. The following steps need to be followed:

• Download MongoDB server jar from here
• Run MongoDB server. Look up the exact instruction in MongoDB site. The general format is as follows:

<path to server>/bin/mongod --dbpath <dbpath>/data/db

Running MongoDB persistence test cases

Once the server is running, the test cases for MongoDB persistence can be run as follows:

• Set up $AKKA_HOME
• Run the following test cases from $AKKA_HOME/akka-persistence

mvn -Dtest=se.scalablesolutions.akka.state.MongoPersistentActorSpec test
mvn -Dtest=se.scalablesolutions.akka.state.MongoStorageSpec test

Redis

Akka also supports Redis as the persistent store for transactors. Redis is an advanced key-value store, actually it's called a data structure server. It is similar to memcached but the dataset is not volatile, and values can be strings, exactly like in memcached, but also lists, sets, and ordered sets.

Create Redis based persistent datastructures

val map =    RedisStorage.newMap(id)
val vector = RedisStorage.newVector(id)
val ref =    RedisStorage.newRef(id)

Redis configuration

Akka’s Redis configuration is included as part of $AKKA_HOME/config/akka.conf. Please refer to the following section in the configuration file:

<storage>
  <redis>
    service = on
    hostname = "127.0.0.1"             # IP address or hostname of the Redis DB instance
    port = 6379                        # default port
  </redis>
</storage>

Running Redis server

Redis server has to be run in order to use Redis as the persistent storage for Akka. The following steps need to be followed:

• Download Redis from here
• Run redis-server. For details on how to set up redis server have alook here

<path to server>./redis-server

Running Redis persistence test cases

Once the server is running, the test cases for Redis persistence can be run as follows:

• Set up $AKKA_HOME
• Run the following test cases from $AKKA_HOME/akka-persistence-redis

mvn test