ReduceByKeyOperator extends OneStreamStateHandlerOperator { @Inject private ReduceByKeyOperator( @Parameter(KeyIndex.class) final int keyIndex, @Parameter(SerializedUdf.class) final String serializedObject, final ClassLoader classLoader) throws IOException, ClassNotFoundException { this(keyIndex, SerializeUtils.deserializeFromString(serializedObject, classLoader)); } @Inject private  ReduceByKeyOperator(
      @Parameter(KeyIndex.class) final int keyIndex,
      @Parameter(SerializedUdf.class) final String serializedObject,
      final ClassLoader classLoader); @Inject  ReduceByKeyOperator(@Parameter(KeyIndex.class) final int keyIndex,
                             final MISTBiFunction<V, V, V> reduceFunc); @Override void processLeftData(final MistDataEvent input); @Override void processLeftWatermark(final MistWatermarkEvent input); @Override Map<String, Object> getStateSnapshot(); @SuppressWarnings("unchecked") @Override void setState(final Map<String, Object> loadedState); @Override void processLeftCheckpoint(final MistCheckpointEvent input);  }
@Test public void testReduceByKeyOperator() throws InjectionException { final List<MistDataEvent> inputStream = ImmutableList.of( createTupleEvent("a", 1, 1L), createTupleEvent("b", 1, 2L), createTupleEvent("c", 1, 3L), createTupleEvent("a", 1, 4L), createTupleEvent("d", 1, 5L), createTupleEvent("a", 1, 6L), createTupleEvent("b", 1, 7L)); final Map<String, Integer> o0 = new HashMap<>(); o0.put("a", 1); final Map<String, Integer> o1 = new HashMap<>(o0); o1.put("b", 1); final Map<String, Integer> o2 = new HashMap<>(o1); o2.put("c", 1); final Map<String, Integer> o3 = new HashMap<>(o2); o3.put("a", 2); final Map<String, Integer> o4 = new HashMap<>(o3); o4.put("d", 1); final Map<String, Integer> o5 = new HashMap<>(o4); o5.put("a", 3); final Map<String, Integer> o6 = new HashMap<>(o5); o6.put("b", 2); final List<MistEvent> expectedStream = ImmutableList.of( new MistDataEvent(o0, 1L), new MistDataEvent(o1, 2L), new MistDataEvent(o2, 3L), new MistDataEvent(o3, 4L), new MistDataEvent(o4, 5L), new MistDataEvent(o5, 6L), new MistDataEvent(o6, 7L)); final int keyIndex = 0; final MISTBiFunction<Integer, Integer, Integer> wordCountFunc = (oldVal, val) -> oldVal + val; final ReduceByKeyOperator<String, Integer> wcOperator = new ReduceByKeyOperator<>(keyIndex, wordCountFunc); final List<MistEvent> result = new LinkedList<>(); wcOperator.setOutputEmitter(new OutputBufferEmitter(result)); inputStream.stream().forEach(wcOperator::processLeftData); LOG.info("expected: " + expectedStream); LOG.info("result: " + result); Assert.assertEquals(expectedStream, result); }