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

EsperBoltesperBolt = newEsperBolt.Builder()
.inputs()
.aliasComponent("ZeroSecondCallBolt")
.withFields("a", "b")
.ofType(Integer.class)
.toEventType("CALL_END")
.outputs()
.outputs().onDefaultStream().emit("count")
.statements()
.add("select callID as CALL_ID,callType as CALL_TYPE,
count(*) as OCCURRENCE_CNT from CDR.win:time_batch(5
minutes) where (eventType = 'CALL_END') and (duration =
0) group by callID,eventType having count(*) > 0 order by
OCCURRENCE_CNTdesc")
.build();
Here is what the output would be like:
The Esper query in the preceding figure executes on a stream of incoming data; here is its
breakdown and explanation:
selectcallID as CALL_ID,callType as CALL_TYPE, count(*) as
OCCURRENCE_CNT
We are selecting the following fields from the incoming tuples, such as Call_Id ,
Call_type , and count :
fromCDR.win:time_batch(5 minutes) where (eventType =
'CALL_END') and (duration = 0) group by
callID,eventTypehaving count(*) > 0
order by OCCURRENCE_CNTdesc