Adding NoSQL Persistence to Storm - Real-time Analytics with Storm and Cassandra - page 140

String nodes =
"10.3.1.41,10.3.1.42,10.3.1.44,10.3.1.45";
String clusterName = "mycluster";
//creating a new configurator
CassandraHostConfigurator hostConfigurator =
new CassandraHostConfigurator(nodes);
hostConfigurator.setCassandraThriftSocketTimeout(0);
cluster =
HFactory.getOrCreateCluster(clusterName,
hostConfigurator);
String[] nodeList = nodes.split(",");
if (nodeList != null && nodeList.length ==
cluster.getConnectionManager().
getDownedHosts().size()) {
logger.error("All cassandra nodes are down. "
+ nodes);
}
//setting up read and write consistencies
ConfigurableConsistencyLevel consistency = new
ConfigurableConsistencyLevel();
consistency.setDefaultWriteConsistencyLevel
(HConsistencyLevel.ONE);
consistency.setDefaultReadConsistencyLevel
(HConsistencyLevel.ONE);
keySpaceObj = HFactory.createKeyspace
("my_keyspace", cluster, consistency);
stringMutator =
HFactory.createMutator(keySpaceObj, stringSerializer);
uuidMutator = HFactory.createMutator
(keySpaceObj, timeUUIDSerializer);
logger.info("Cassandra data store initialized,
Nodes=" + nodes + ", " + "cluster name=" +
clusterName + ", " + "keyspace=" + keyspace + ", " +
"consistency=" + writeConsistency);
}