DistinctMap extends BaseKeyValueOperator<K, V> { @Override public void endWindow() { mapkeyval.clear(); }  @Override void endWindow(); final transient DefaultInputPort<Map<K, V>> data; final transient DefaultOutputPort<HashMap<K, V>> distinct; }
@SuppressWarnings({ "rawtypes", "unchecked" }) @Test public void testNodeProcessing() throws Exception { DistinctMap<String, Number> oper = new DistinctMap<String, Number>(); CollectorTestSink sortSink = new CollectorTestSink(); oper.distinct.setSink(sortSink); oper.beginWindow(0); HashMap<String, Number> input = new HashMap<String, Number>(); input.put("a", 2); oper.data.process(input); input.clear(); input.put("a", 2); oper.data.process(input); input.clear(); input.put("a", 1000); oper.data.process(input); input.clear(); input.put("a", 5); oper.data.process(input); input.clear(); input.put("a", 2); input.put("b", 33); oper.data.process(input); input.clear(); input.put("a", 33); input.put("b", 34); oper.data.process(input); input.clear(); input.put("b", 34); oper.data.process(input); input.clear(); input.put("b", 6); input.put("a", 2); oper.data.process(input); input.clear(); input.put("c", 9); oper.data.process(input); oper.endWindow(); Assert.assertEquals("number emitted tuples", 8, sortSink.collectedTuples.size()); int aval = 0; int bval = 0; int cval = 0; for (Object o: sortSink.collectedTuples) { for (Map.Entry<String, Integer> e: ((HashMap<String, Integer>)o).entrySet()) { String key = e.getKey(); if (key.equals("a")) { aval += e.getValue(); } else if (key.equals("b")) { bval += e.getValue(); } else if (key.equals("c")) { cval += e.getValue(); } } } Assert.assertEquals("Total for key \"a\" ", 1040, aval); Assert.assertEquals("Total for key \"a\" ", 73, bval); Assert.assertEquals("Total for key \"a\" ", 9, cval); }