SDN Controller Programming Guide

94
In addition, there is need to create/update the secondary column families to keep the queries
updated. The above mentioned interface operations provide an abstraction to perform a write on
all secondary column families along with the main column family.
The secondary column family needs to define the necessary serializers for composite columns and
a RowKey. In the demo code, every secondary column family has exactly one very wide row. This
is done to achieve faster lookup during a read operation. If the data exceeds the upper limit of the
number of columns (2 billion columns), other methods such as sharding can be used to partition
the secondary index.
In the example, AlertsBySeverity is shown. Similar code needs to be written for each secondary
column family that is needed by the query operations of the application.
Once all secondary column families are defined along with main column family, the DAO needs
to provide the following methods. The example code of these methods as defined in the demo
application is presented here.
convert()
This method is used during read operations. When a row needs to be returned to the application,
it converts the data from storable format to a DTO.
CassandraAlertDao.java:
@Override
public CassandraAlert convert(CassandraStorable<String, String> source) {
if (source == null) {
throw new NullPointerException(...);
}
final CassandraAlert alert = new CassandraAlert(source.getId());
ColumnVisitor<String> visitor = new ColumnVisitorAdapter<String>() {
@Override
public void visit(BooleanColumn<String> column) {
alert.setState(column.getValue());
}
@Override
public void visit(DateColumn<String> column) {
alert.setTimestamp(column.getValue());
}
@Override
public void visit(StringColumn<String> column) {
if (AlertColumnFamily.DESC_COL_NAME.equals(column.getName())) {
alert.setDescription(column.getValue());
} else if (AlertColumnFamily.ORIGIN_COL_NAME
.equals(column.getName())) {
alert.setOrigin(column.getValue());
} else if (AlertColumnFamily.TOPIC_COL_NAME
.equals(column.getName())) {