Titan Server: From a Single Server to a Highly Available Cluster

titan is a distributed graph database capable of storing graphs on the order of hundreds of billions of edges while, at the same time, supporting billions of real-time graph traversals a day. for most graph applications, the high-end performance aspects of titan will never be reached. this does not mean that titan is unsuitable for graph applications at the smaller scale — in the billions of edges and below. the purpose of this post is to introduce titan from the perspective of a team of engineers developing a new graph-based application. these engineers will initially develop and test their codebase using a single titan server . when the application matures and is ready for production use, a highly-available setup is deployed. finally, as the application becomes more popular and the data size and transactional load increases, a fully distributed cluster is leveraged. growing a titan database from a single server to a cluster is simply a matter of configuration. in this way, titan gracefully scales to accommodate the changing requirements of a graph application.

titan single server

augustus and tiberius are two software engineers who have designed an application that represents the gods and people of rome within a graph of familial relationships — a genealogy application. the intention is that roman scholars will use the application to better understand the social fabric of their great empire. while the intention is single-user, the two engineers decide to leverage titan as the backend graph database. for one, titan is completely free for any use ( apache 2 licensed ) and two, it supports both single server and distributed deployments. the latter is important to them because the greek oracle of delphi foretold that a genealogy graph would one day be used online by everyone throughout the roman empire .

$ wget http://s3.thinkaurelius.com/downloads/titan/titan-cassandra-0.3.0.zip
$ unzip titan-cassandra-0.3.0.zip
$ cd titan-cassandra-0.3.0
$ sudo bin/titan.sh config/titan-server-rexster.xml config/titan-server-cassandra.properties
13/03/27 12:40:32 info service.cassandradaemon: jvm vendor/version: java hotspot(tm) 64-bit server vm/1.7.0_12-ea
13/03/27 12:40:32 info service.cassandradaemon: heap size: 40566784/477233152
13/03/27 12:40:32 info config.databasedescriptor: loading settings from file:/users/marko/software/aurelius/titan/config/cassandra.yaml
13/03/27 12:40:32 info config.databasedescriptor: global memtable threshold is enabled at 151mb
13/03/27 12:40:32 info service.cacheservice: initializing key cache with capacity of 2 mbs.
13/03/27 12:40:35 info server.rexprorexsterserver: rexpro serving on port: [8184]
13/03/27 12:40:35 info server.httprexsterserver: rexster server running on: [http://localhost:8182]
13/03/27 12:40:35 info server.shutdownmanager: bound shutdown socket to /127.0.0.1:8183. starting listener thread for shutdown requests.
...

    users without wget can use curl -o or download from the titan download page. 

the above sequence of 4 shell commands downloads and starts up a titan server on the localhost . titan server embeds both cassandra and (a lightweight version of) rexster within the same jvm . titan server exposes the following language-agnostic endpoints for developers to communicate with the graph:

1. a restful endpoint available at http://localhost:8182/graphs .
   • crud operations can be enacted via http and json .
2. a rexpro binary protocol endpoint available on port 8184 .
   • arbitrarily complex gremlin read/write operations can be enacted via messagepack .

titan server is configured via two primary files: titan-server-rexster.xml (shown below) and cassandra.yaml (discussed in the next section). these files are located in the config/ directory of the titan-cassandra-x.y.z distribution.

<rexster>
    <http>
        <server-port>8182</server-port>
        <server-host>0.0.0.0</server-host>
        <base-uri>http://localhost</base-uri>
        <character-set>utf-8</character-set>
        <enable-jmx>false</enable-jmx>
        <max-post-size>2097152</max-post-size>
        <max-header-size>8192</max-header-size>
        <upload-timeout-millis>30000</upload-timeout-millis>
        <thread-pool>
            <worker>
                <core-size>8</core-size>
                <max-size>8</max-size>
            </worker>
            <kernal>
                <core-size>4</core-size>
                <max-size>4</max-size>
            </kernal>
        </thread-pool>
        <io-strategy>leader-follower</io-strategy>
    </http>
    <rexpro>
        <server-port>8184</server-port>
        <server-host>0.0.0.0</server-host>
        <session-max-idle>1790000</session-max-idle>
        <session-check-interval>3000000</session-check-interval>
        <connection-max-idle>180000</connection-max-idle>
        <connection-check-interval>3000000</connection-check-interval>
        <enable-jmx>false</enable-jmx>
        <thread-pool>
            <worker>
                <core-size>8</core-size>
                <max-size>8</max-size>
            </worker>
            <kernal>
                <core-size>4</core-size>
                <max-size>4</max-size>
            </kernal>
        </thread-pool>
        <io-strategy>leader-follower</io-strategy>
    </rexpro>
    <security>
        <authentication>
            <type>none</type>
        </authentication>
    </security>
    <shutdown-port>8183</shutdown-port>
    <shutdown-host>127.0.0.1</shutdown-host>
</rexster>

note : along with the above endpoints, titan server also exposes a jvm native serialization interface that can be used by jvm languages. this interface, for example, is the means by which faunus / hadoop interacts with titan server for global graph analytics. for more information on this endpoint, see using cassandra .

titan highly available

the genealogy application was showing promise as a single-user system for studying the genetic history of the roman people and gods. due to the positive response, augustus and tiberius decide that a multi-user online genealogy service would be a successful product.

 how many siblings did jupiter have?
g.v('name','jupiter').out('brother','sister').count()
// who is caesar's grandmother?
g.v('name','caesar').out('mother').out('mother').name
// who are marcus' children's in-laws?
g.v('name','marcus').in('father').has('gender','m').out('married').out('father','mother').name 

as it currently stands, the genealogy data set is approximately 1 billion edges. therefore, it can be stored and processed on a single machine. as a single-user application a single titan server suffices. however, with multiple users, it is important that the system is robust and can serve numerous concurrent requests. if the application is only backed by a single server, then if that server goes down, the application is unusable. to ensure 1.) no single point of failure and 2.) support for more transactions per second, augustus and tiberius deploy 3 machines each with a titan server installed.

the team updates the default config/cassandra.yaml file of each titan server by changing the localhost property value to be the ip address of the machine and adding a seed ip address for discoverability (see multinode cluster ). next, they start each titan server one after the other ( titan.sh ). to ensure that the servers properly clustered together, they use cassandra’s nodetool .

apache-cassandra-1.2.3$ bin/nodetool ring
 
datacenter: datacenter1
==========
replicas: 1
 
address         rack        status state   load            owns                token
                                                                               57715295010532946864463892271081778854
10.223.14.57    rack1       up     normal  93.06 kb        49.28%              141555886663081320436455748965948652071
10.174.123.131  rack1       up     normal  59.73 kb        33.44%              28311611028231080169766921879398209884
10.196.0.207    rack1       up     normal  9.43 kb         17.28%              57715295010532946864463892271081778854

finally, on one of the servers, the cassandra-cli tool is used to update the replication factor of the titan -keyspace.

apache-cassandra-1.2.3$ bin/cassandra-cli -h 10.174.123.131
 
[default@unknown] update keyspace titan with strategy_options = {replication_factor:3};
a3b7e1a3-4a88-3769-8c5e-90cda4fec0e1
[default@unknown] show schema titan;
create keyspace titan
  with placement_strategy = 'simplestrategy'
  and strategy_options = {replication_factor : 3}
  and durable_writes = true;

with a replication factor of 3, each of the 3 titan servers are the primary host of approximately one-third of the vertices in the graph while, at the same time, each maintains a replica of the primary data of the other two servers. in this way, a highly-available, master-master setup is rendered. with this model, there is no single point of failure. if one of the database machines goes down, the other two are able to serve the primary data of the dead machine. if two of the machines go down, the remaining machine can serve data — albeit not with the same throughput possible when all three machines are available. with full master-master replication, the graph is duplicated and each server can support both reads and writes to the graph.

titan clustered

the following summer, the prophecy of the oracle of delphi comes true. an announcement is made in the roman forum about the utility of the online genealogy application. immediately, the plebeians of rome join the site. they feverishly add their family histories and traverse the graph to learn more about their genetic past. this spike in usage puts an excessive amount of strain on the servers. with so many concurrent users, the three server machines have their cpu and disk i/o peaked trying to process requests.

to remedy the situation, 6 more titan server machines are added to the cluster for a total of 9 machines. the token ring is rebalanced to ensure that each server maintains a relatively equal amount of the graph. a perfect/fair partition of 2^128 into 9 parts is below (see token ring calculator ).

0
18904575940052135809661593108510408704
37809151880104271619323186217020817408
56713727820156407428984779325531226112
75618303760208543238646372434041634816
94522879700260688493040931281842470912
113427455640312814857969558651062452224
132332031580364960112364117498863288320
151236607520417086477292744868083269632

each machine has its token updated using the following nodetool command. by repartitioning the token ring, the 3 original servers transfer their data to the newly on-boarded servers in order to distributed the data load as specified by their location in the 128-bit token space (each vertex hashes to a particular 128-bit token).

apache-cassandra-1.2.3$ bin/nodetool -h 10.223.14.57 move 0
apache-cassandra-1.2.3$ bin/nodetool -h 10.174.123.131 move 18904575940052135809661593108510408704
apache-cassandra-1.2.3$ bin/nodetool -h 10.196.0.207 move 37809151880104271619323186217020817408
...
..
.

with the replication factor still set to 3, each server does not maintain a full replica of the graph. instead, each server only replicates a third of the full graph (3/9). at this point, no single server has a full picture of the graph. however, because there are more servers, more transactions can be served and more data can be stored. augustus and tiberius have successfully grown their single-user graph application to a distributed system that stores and processes a massive genealogy graph represented across a cluster of titan server machines.

conclusion

titan was developed from the outset to support oltp distributed graph storage and processing. while it is important that a graph database can scale indefinitely, less than 1% of applications written today will ever leverage near trillion edge graphs. the other 99% of applications will store and process million and billion edge graphs. titan is able to meet the requirements of both segments of the graph application space. furthermore, titan scales gracefully as developers move from a single server prototype, to a highly-available production system, to ultimately, a fully distributed cluster sustaining the size and workload requirements seen by 1% of applications.

acknowledgements

stephen mallette and blake eggleston are the developers of rexster’s rexpro . their efforts were a driving force behind the development of titan server .