Processing Streaming Data - Real-Time Analytics - page 176

TridentTopology topology = new TridentTopology();
topology.newStream("lorem", new LoremIpsumSpout()
.tuple("first","second","third"))
.each( new Fields(), new RandomFilter())
;
Note the empty Fields constructor. This is because the RandomFilter
does not depend on any fields to evaluate the filter. It does not actually
modify the Tuple passed through, which still contains the “first” ,
“second” , and “third” elements.
Similarly, a function's each argument takes one of two forms. The first
is each(function,outputFields) , the form used by the
RandomFunction implementation:
TridentTopology topology = new TridentTopology();
topology.newStream("lorem", new LoremIpsumSpout()
.tuple("first","second","third"))
.each( new RandomFunction(), new Fields("x"))
;
The second form allows the function to take input fields, the same as a filter
operation. For example, a SquareFunction implementation would take a
single parameter:
public class SquareFunction extends BaseFunction {
private static final long serialVersionUID = 1L;
public void execute(TridentTuple tuple,
TridentCollector collector) {
collector.emit( new
Values(tuple.getDouble(0)*tuple.getDouble(0)));
}
}
When assembled into a topology with the RandomFunction , it would take
the x field as its input and produce a y field as output that would be
appended to the tuple: