storm-cassandra

通过一个动态可配置的

backtype.storm.Bolt

实现将Storm和Cassandra集成在一起，在类Class中指定column family, row key, and column name/values，则Bolt将Storm

Tuple

对象写入Cassandra cluster的Column Family。

具体操作：

Basic Usage

CassandraBolt

TridentCassandraLookupFunction

TridentCassandraWriteFunction

 

均以key/map形式调用Storm topology配置中为Cassandra指定的hostname、port、keyspace，以允许topology中的多个实例可以连接同一个Cassandra实例。key在构造函数中指定以表示要使用的map：

Map<String, Object> cassandraConfig = new HashMap<String, Object>();
cassandraConfig.put(StormCassandraConstants.CASSANDRA_HOST, "localhost:9160");
cassandraConfig.put(StormCassandraConstants.CASSANDRA_KEYSPACE, "testKeyspace");
Config config = new Config();
config.put("CassandraLocal", cassandraConfig);

CassandraBolt

的构造器将column family name和 row key field的值作为参数：

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

上述创建的

CassandraBolt

写入名为"

columnFamily

" 的column family，并且查询/使用

backtype.storm.tuple.Tuple

对象中的"

rowKey

"作为 Cassandra row key。

backtype.storm.Tuple

接受的每一field，

CassandraBolt

将写入name/value对。比如，上述的构造器如下：

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

则Cassandra row如下 ( from

cassandra-cli

):

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

Cassandra Write Function

如果Function不会emit任何数据，则Storm Trident 过滤掉original Tuple。在写入Cassandra之后，

TridentCassandraWriteFunction

则emit一个static Object值以进行其他处理，此主要使Tuple继续，而static value或由构造器或调用setValueToEmitAfterWrite方法设置，而如果设置为NULL则使得此Function不能emit而且Storm将过滤掉此Tuple。默认的行为是不emit，如果Function
emit 一个值，不要忘记 声明输出域。

Cassandra Counter Columns

The Counter Column与上相似，必须具体化rowKey 和value来具体化计数。所有其他field将按此计数累加。

CassandraCounterBatchingBolt logPersistenceBolt = new CassandraCounterBatchingBolt(
"columnFamily", "RowKeyField", "IncrementAmountField" );

上述创建的bolt 写入名为"

columnFamily

"的 column family并用名为 "

RowKeyField

"指定需要接受的Tuple。 此Tuple的其他field则隐含增加

IncrementAmountField

指定的值。

给定如下Tuple:

{rowKey: 12345, IncrementAmount: 1L, IncrementColumn: 'SomeCounter'}

将为 "

SomeCounter

" 增加 1L。