AverageKeyVal extends BaseNumberKeyValueOperator<K, Number> { @Override public void endWindow() { for (Map.Entry<K, MutableDouble> e : sums.entrySet()) { K key = e.getKey(); double d = e.getValue().doubleValue(); if (doubleAverage.isConnected()) { doubleAverage.emit(new KeyValPair<K, Double>(key, d / counts.get(key).doubleValue())); } if (intAverage.isConnected()) { intAverage.emit(new KeyValPair<K, Integer>(key, (int)d)); } if (longAverage.isConnected()) { longAverage.emit(new KeyValPair<K, Long>(key, (long)d)); } } sums.clear(); counts.clear(); }  @Override void endWindow(); final transient DefaultInputPort<KeyValPair<K, ? extends Number>> data; @OutputPortFieldAnnotation(optional = true)
  final transient DefaultOutputPort<KeyValPair<K, Double>> doubleAverage; @OutputPortFieldAnnotation(optional = true)
  final transient DefaultOutputPort<KeyValPair<K, Integer>> intAverage; @OutputPortFieldAnnotation(optional = true)
  final transient DefaultOutputPort<KeyValPair<K, Long>> longAverage; }
@SuppressWarnings({ "unchecked", "rawtypes" }) @Test public void testNodeProcessing() { AverageKeyVal<String> oper = new AverageKeyVal<String>(); CollectorTestSink averageSink = new CollectorTestSink(); oper.doubleAverage.setSink(averageSink); oper.beginWindow(0); oper.data.process(new KeyValPair("a", 2.0)); oper.data.process(new KeyValPair("b", 20.0)); oper.data.process(new KeyValPair("c", 1000.0)); oper.data.process(new KeyValPair("a", 1.0)); oper.data.process(new KeyValPair("a", 10.0)); oper.data.process(new KeyValPair("b", 5.0)); oper.data.process(new KeyValPair("d", 55.0)); oper.data.process(new KeyValPair("b", 12.0)); oper.data.process(new KeyValPair("d", 22.0)); oper.data.process(new KeyValPair("d", 14.2)); oper.data.process(new KeyValPair("d", 46.0)); oper.data.process(new KeyValPair("e", 2.0)); oper.data.process(new KeyValPair("a", 23.0)); oper.data.process(new KeyValPair("d", 4.0)); oper.endWindow(); Assert.assertEquals("number emitted tuples", 5, averageSink.collectedTuples.size()); for (Object o : averageSink.collectedTuples) { KeyValPair<String, Double> e = (KeyValPair<String, Double>)o; Double val = e.getValue(); if (e.getKey().equals("a")) { Assert.assertEquals("emitted value for 'a' was ", new Double(36 / 4.0), val); } else if (e.getKey().equals("b")) { Assert.assertEquals("emitted tuple for 'b' was ", new Double(37 / 3.0), val); } else if (e.getKey().equals("c")) { Assert.assertEquals("emitted tuple for 'c' was ", new Double(1000 / 1.0), val); } else if (e.getKey().equals("d")) { Assert.assertEquals("emitted tuple for 'd' was ", new Double(141.2 / 5), val); } else if (e.getKey().equals("e")) { Assert.assertEquals("emitted tuple for 'e' was ", new Double(2 / 1.0), val); } } }