Hi all,
I've been trying to set up a really simple Trident topology to try
*persistentAggregate* with a ReducerAggregator based on the examples found
in the docs and I might have got something wrong because my topology fails
with ArrayIndexOutOfBoundsException when trying to retrieve a value from a
tuple at the *reduce* function.
Caused by: java.lang.ArrayIndexOutOfBoundsException: 0
at storm.trident.tuple.TridentTupleView.getValue(TridentTupleView.java:258)
~[storm-core-0.11.0-SNAPSHOT.jar:0.11.0-SNAPSHOT]
at
hu.sztaki.state.machine.paper.trident.DummyTridentTopology$1.reduce(DummyTridentTopology.java:44)
~[classes/:?]
at
hu.sztaki.state.machine.paper.trident.DummyTridentTopology$1.reduce(DummyTridentTopology.java:38)
~[classes/:?]
at
storm.trident.state.ReducerValueUpdater.update(ReducerValueUpdater.java:37)
~[storm-core-0.11.0-SNAPSHOT.jar:0.11.0-SNAPSHOT]
my code is basically:
topology
.newStream("simpleSpout", new FixedBatchSpout(new Fields("A", "B"), 3,
new Values(1, 1),
new Values(2, 1),
...
))
.persistentAggregate(
new MemoryMapState.Factory(),
new ReducerAggregator<String>() {
public String init() {
return "empty";
}
public String reduce(String curr, TridentTuple tuple) {
* return curr + ":" + tuple.getValue(0);*
}
}, new Fields("C"));
The cause of this is an empty tuple at the reduce function, but I don't
understand why is that. Should I declare somewhere the input fields of the
persistentAggregator?
Could you please help me figuring out this?
I am using the current master branch (version 0.11.0-SNAPSHOT) but the
0.9.5 release also produces this error.
My full code can be found here <http://pastebin.com/dcbDjte8> and the
output of the execution here <http://pastebin.com/jgYfGzAZ>.
(As a note: Wouldn't it would be more convenient to throw an exception with
a more appropriate text like "No such field in tuple" to make the errors
easier to understand.)
Thanks for your help in advance.
Best regards,
Gabor Hermann
