UniqueValueCount extends BaseOperator { @Override public void endWindow() { for (K key : interimUniqueValues.keySet()) { Set<Object> values = interimUniqueValues.get(key); if (output.isConnected()) { output.emit(new InternalCountOutput<>(key, values.size(), values)); } if (outputValues.isConnected()) { outputValues.emit(new KeyValPair<>(key, values)); } } interimUniqueValues.clear(); } UniqueValueCount(); @Override void endWindow(); transient DefaultInputPort<KeyValPair<K, ? extends Object>> input; transient DefaultOutputPort<KeyValPair<K, Integer>> output; final transient DefaultOutputPort<KeyValPair<K, Set<Object>>> outputValues; }
@Test public void uniqueCountTest() { UniqueValueCount<String> uniqueCountOper = new UniqueValueCount<String>(); CollectorTestSink<KeyValPair<String, Integer>> outputSink = new CollectorTestSink<KeyValPair<String, Integer>>(); CollectorTestSink<KeyValPair<String, Set<Object>>> outputSetSink = new CollectorTestSink<KeyValPair<String, Set<Object>>>(); TestUtils.setSink(uniqueCountOper.output, outputSink); TestUtils.setSink(uniqueCountOper.outputValues, outputSetSink); uniqueCountOper.beginWindow(0); uniqueCountOper.input.process(new KeyValPair<String, Object>("test1", 1)); uniqueCountOper.input.process(new KeyValPair<String, Object>("test1", 2)); uniqueCountOper.input.process(new KeyValPair<String, Object>("test1", 2)); uniqueCountOper.endWindow(); Assert.assertEquals("number emitted tuples", 1, outputSink.collectedTuples.size()); KeyValPair<String, Integer> emittedPair = outputSink.collectedTuples.get(0); Assert.assertEquals("emitted key was ", "test1", emittedPair.getKey()); Assert.assertEquals("emitted value was ", 2, emittedPair.getValue().intValue()); Assert.assertEquals("number emitted tuples", 1, outputSetSink.collectedTuples.size()); KeyValPair<String, Set<Object>> emittedSetPair = outputSetSink.collectedTuples.get(0); Assert.assertTrue(emittedSetPair.getValue().contains(1)); Assert.assertTrue(emittedSetPair.getValue().contains(2)); outputSink.clear(); uniqueCountOper.beginWindow(1); uniqueCountOper.input.process(new KeyValPair<String, Object>("test1", 1)); uniqueCountOper.input.process(new KeyValPair<String, Object>("test1", 2)); uniqueCountOper.input.process(new KeyValPair<String, Object>("test1", 2)); uniqueCountOper.input.process(new KeyValPair<String, Object>("test2", 1)); uniqueCountOper.input.process(new KeyValPair<String, Object>("test2", 2)); uniqueCountOper.input.process(new KeyValPair<String, Object>("test2", 2)); uniqueCountOper.endWindow(); ImmutableMap<String, Integer> answers = ImmutableMap.of("test1", 2, "test2", 2); Assert.assertEquals("number emitted tuples", 2, outputSink.collectedTuples.size()); for (KeyValPair<String, Integer> emittedPair2 : outputSink.collectedTuples) { Assert.assertEquals("emmit value of " + emittedPair2.getKey() + " was ", answers.get(emittedPair2.getKey()), emittedPair2.getValue()); } LOG.debug("Done unique count testing testing\n"); }