Steve, potentially the input into your counter bolt is malformed given your
Cassandra schema? Have you had success writing into Cassandra in the past
with this setup and Storm?
So, I'm very new to Storm, but I've gotten Storm-Cassandra to work in my
setup. For reference, the package dependency I'm using is Maven
<dependency><groupId>com.hmsonline</groupId><artifactId>storm-cassandra</artifactId><version>0.4.0-rc4</version></dependency>.
The version of Cassandra 2.0.4 is DataStax dsc20-2.0.4-1
cassandra20-2.0.4-1 with Oracle Java 7u51 x64, Storm 0.9.0.1, and Zk 3.4.5.
Cassandra schema: CREATE KEYSPACE stormks; CREATE COLUMN FAMILY mycf WITH
default_validation_class=CounterColumnType AND
key_validation_class=AsciiType AND comparator=AsciiType;
I have had success writing into Cassandra by creating my own
CassandraWriterBolt class, which ultimately extends the Astyanax client
through the storm-cassandra package's CassandraBolt. There are most likely
billions of ways to do this better, but it works for me thus far, so I've
attached the the class and implementation for reference to this email
thread. Example usage for your topology:
String cassandra_host = "localhost:9160";
String cassandra_config_key = "cassandra-config";
String cassandra_key_space = "stormks";
String cassandra_cf = "mycf";
String cassandra_name_key = "key";
String cassandra_name_inc_amount = "inc_amount";
CassandraWriterBolt<String,String,String> cassandraWriterBolt =
new CassandraWriterBolt<>(
cassandra_host, cassandra_config_key,
cassandra_key_space,
cassandra_cf, cassandra_name_key,
cassandra_name_inc_amount
);
/***/
conf.put(cassandraWriterBolt.getClientConfigKey(),
cassandraWriterBolt.getClientConfig());
For the setup bolt which forms the output tuple to stream to the
CassandraWriterBolt implementation, you just need to keep consistent the
expectations set for the CassandraWriterBolt on instantiation, namely the
cassandra_name_key and the cassandra_name_inc_amount naming conventions.
In example; I'd send a Tuple ~
("myCasskey",1L,"aCassColumnNameWCounter","bCassColumnNameWCounter",...etc.)
and with declareOutputFields' declarer containing new Fields(new
String[]{"key","inc_amount","column_a","column_b",...etc.}), which should
update Cassandra's stormks::mycf: myCassKey[aCassColumnNameWCounter] += 1L,
myCassKey[bCassColumnNameWCounter] += 1L.
Hope this helps towards fixing your issue,
-Dan
package com.civicscience.storm.weblog.bolt;
import backtype.storm.task.OutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.IRichBolt;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.tuple.Tuple;
import java.util.Arrays;
import java.util.HashMap;
import java.util.Map;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
* CassandraWriterBolt.java
*
* @param <K>
* @param <C>
* @param <V>
* @author Dan DeCapria, Copyright (c) 2011-2014
* @since 05 February 2014
* @version 0.1, 05 February 2014
*/
public class CassandraWriterBolt<K, C, V> extends com.hmsonline.storm.cassandra.bolt.CassandraBolt<K, C, V> implements IRichBolt {
private static final Logger logger = LoggerFactory.getLogger(CassandraWriterBolt.class);
private com.hmsonline.storm.cassandra.bolt.AckStrategy ackStrategy = com.hmsonline.storm.cassandra.bolt.AckStrategy.ACK_ON_WRITE;
private TopologyContext context;
private OutputCollector collector;
private String cassandraHost;
private String clientConfigKey;
private String keyspace;
private com.hmsonline.storm.cassandra.bolt.mapper.TupleCounterMapper tupleCounterMapper;
private HashMap<String, Object> clientConfig;
private CassandraWriterBolt(String clientConfigKey, com.hmsonline.storm.cassandra.bolt.mapper.TupleMapper<K, C, V> tupleMapper) {
super(clientConfigKey, tupleMapper);
this.clientConfigKey = clientConfigKey;
}
/**
* CivicScience Implementation Constructor.
* @param cassandraHost
* @param clientConfigKey
* @param keyspace
* @param columnFamily
* @param rowKeyField
* @param incrementAmountField
*/
public CassandraWriterBolt(String cassandraHost, String clientConfigKey, String keyspace, String columnFamily, String rowKeyField, String incrementAmountField) {
super(clientConfigKey, null);
this.cassandraHost = cassandraHost;
this.clientConfigKey = clientConfigKey;
this.keyspace = keyspace;
this.tupleCounterMapper = new com.hmsonline.storm.cassandra.bolt.mapper.DefaultTupleCounterMapper(this.keyspace, columnFamily, rowKeyField, incrementAmountField);
this.clientConfig = new HashMap<>();
this.clientConfig.put(com.hmsonline.storm.cassandra.StormCassandraConstants.CASSANDRA_HOST, this.cassandraHost);
this.clientConfig.put(com.hmsonline.storm.cassandra.StormCassandraConstants.CASSANDRA_KEYSPACE, Arrays.asList(new String[]{this.keyspace}));
}
@Override
public void declareOutputFields(OutputFieldsDeclarer declarer) {
// we don't emit anything from here.
}
@Override
public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) {
this.context = context;
this.collector = collector;
super.prepare(stormConf, this.context);
}
@Override
public void execute(Tuple input) {
try {
this.incrementCounter(input, this.tupleCounterMapper);
if (this.ackStrategy == com.hmsonline.storm.cassandra.bolt.AckStrategy.ACK_ON_WRITE) {
this.collector.ack(input);
}
} catch (Exception ex) {
logger.error("Exception: " + ex.getMessage() + ";\t tuple = " + input);
this.collector.reportError(ex);
this.collector.fail(input);
}
}
@Override
public void cleanup() {
super.cleanup();
}
public com.hmsonline.storm.cassandra.bolt.AckStrategy getAckStrategy() {
return this.ackStrategy;
}
public void setAckStrategy(com.hmsonline.storm.cassandra.bolt.AckStrategy ackStrategy) {
this.ackStrategy = ackStrategy;
}
public HashMap<String, Object> getClientConfig() {
return this.clientConfig;
}
public String getCassandraHost() {
return this.cassandraHost;
}
public String getClientConfigKey() {
return this.clientConfigKey;
}
public String getKeyspace() {
return this.keyspace;
}
}
