Distributed Cache and CEP with Storm - Real-time Analytics with Storm and Cassandra - page 237

}
private void populateEventFromCache(Event event) {
HashMapfetchMap = (HashMap)
cacheReader.get(searchObj.hashCode());
if (null != fetchMap) {
event.setAccountID(Integer.parseInt((String)
fetchMap.get("account_id")));
logger.debug("Populating event" + event + " using cache
" + fetchMap);
} else {
logger.debug("No matching event found in cache.");
}
logger.info("Time to fetch from cache=" +
(System.currentTimeMillis() - t1) + "msec");
}
}
/**
* Declares output streams and tuple fields emitted from
this bolt
*/
@Override
public void declareOutputFields(OutputFieldsDeclarer
declarer) {
String stormStreamName = logStream.getName() + "_" +
eventType;
declarer.declareStream(stormStreamName, new
Fields(stormStreamName));
logger.debug("Topology : " + topology.getTopologyName() +
", Declared output stream : " + stormStreamName + ",
Output field : " + stormStreamName);
}
The preceding code snippet demonstrates a bolt, which reads an event from the stream,
gets some dimensional data from memcache, and emits the enriched bolt to the streams to
the following bolts in the DAG topology.