ModeOperator extends BaseOperator { @Override public void endWindow() { V modeValue = null; int max = 0; for (Map.Entry<V, Integer> entry : values.entrySet()) { if (entry.getValue() > max) { modeValue = entry.getKey(); max = entry.getValue(); } } if (mode != null) { mode.emit(modeValue); } }  @Override void endWindow(); final transient DefaultInputPort<V> data; final transient DefaultOutputPort<V> mode; }
@Test public void testWeightedMean() { ModeOperator<Double> oper = new ModeOperator<Double>(); CollectorTestSink<Double> sink = new CollectorTestSink<Double>(); TestUtils.setSink(oper.mode, sink); oper.setup(null); oper.beginWindow(0); oper.data.process(1.0); oper.data.process(7.0); oper.data.process(3.0); oper.data.process(9.0); oper.data.process(7.0); oper.data.process(7.0); oper.endWindow(); Assert.assertEquals("Must be one tuple in sink", sink.collectedTuples.size(), 1); Assert.assertTrue("Median value", sink.collectedTuples.get(0) == 7.0); }