@Test public void testBuild_Windowing() { Flow flow = Flow.create("TEST"); Dataset<String> dataset = Util.createMockDataset(flow, 2); ReduceByKey.of(dataset) .keyBy(s -> s) .valueBy(s -> 1L) .combineBy(n -> StreamSupport.stream(n.spliterator(), false).mapToLong(Long::new).sum()) .windowBy(Time.of(Duration.ofHours(1))) .output(); ReduceByKey reduce = (ReduceByKey) flow.operators().iterator().next(); assertTrue(reduce.getWindowing() instanceof Time); assertNull(reduce.valueComparator); }
public static <IN> KeyByBuilder<IN> of(Dataset<IN> input) { return new KeyByBuilder<>("ReduceByKey", input); }
ReduceByKey extends StateAwareWindowWiseSingleInputOperator<
        IN, IN, IN, KEY, Pair<KEY, OUT>, W,
        ReduceByKey<IN, KEY, VALUE, OUT, W>> { public static <IN> KeyByBuilder<IN> of(Dataset<IN> input) { return new KeyByBuilder<>("ReduceByKey", input); } }
ReduceByKey extends StateAwareWindowWiseSingleInputOperator<
        IN, IN, IN, KEY, Pair<KEY, OUT>, W,
        ReduceByKey<IN, KEY, VALUE, OUT, W>> { public static <IN> KeyByBuilder<IN> of(Dataset<IN> input) { return new KeyByBuilder<>("ReduceByKey", input); } @SuppressWarnings("unchecked")  ReduceByKey(String name,
              Flow flow,
              Dataset<IN> input,
              UnaryFunction<IN, KEY> keyExtractor,
              @Nullable Class<KEY> keyClass,
              UnaryFunction<IN, VALUE> valueExtractor,
              @Nullable Windowing<IN, W> windowing,
              CombinableReduceFunction<OUT> reducer,
              Set<OutputHint> outputHints);   ReduceByKey(String name,
              Flow flow,
              Dataset<IN> input,
              UnaryFunction<IN, KEY> keyExtractor,
              @Nullable Class<KEY> keyClass,
              UnaryFunction<IN, VALUE> valueExtractor,
              @Nullable Windowing<IN, W> windowing,
              ReduceFunctor<VALUE, OUT> reducer,
              @Nullable BinaryFunction<VALUE, VALUE, Integer> valueComparator,
              Set<OutputHint> outputHints); }
ReduceByKey extends StateAwareWindowWiseSingleInputOperator<
        IN, IN, IN, KEY, Pair<KEY, OUT>, W,
        ReduceByKey<IN, KEY, VALUE, OUT, W>> { public static <IN> KeyByBuilder<IN> of(Dataset<IN> input) { return new KeyByBuilder<>("ReduceByKey", input); } @SuppressWarnings("unchecked")  ReduceByKey(String name,
              Flow flow,
              Dataset<IN> input,
              UnaryFunction<IN, KEY> keyExtractor,
              @Nullable Class<KEY> keyClass,
              UnaryFunction<IN, VALUE> valueExtractor,
              @Nullable Windowing<IN, W> windowing,
              CombinableReduceFunction<OUT> reducer,
              Set<OutputHint> outputHints);   ReduceByKey(String name,
              Flow flow,
              Dataset<IN> input,
              UnaryFunction<IN, KEY> keyExtractor,
              @Nullable Class<KEY> keyClass,
              UnaryFunction<IN, VALUE> valueExtractor,
              @Nullable Windowing<IN, W> windowing,
              ReduceFunctor<VALUE, OUT> reducer,
              @Nullable BinaryFunction<VALUE, VALUE, Integer> valueComparator,
              Set<OutputHint> outputHints); static KeyByBuilder<IN> of(Dataset<IN> input); static OfBuilder named(String name); @Nullable Class<KEY> getKeyClass(); ReduceFunctor<VALUE, OUT> getReducer(); boolean isCombinable(); UnaryFunction<IN, VALUE> getValueExtractor(); @Nullable BinaryFunction<VALUE, VALUE, Integer> getValueComparator(); @SuppressWarnings("unchecked") @Override DAG<Operator<?, ?>> getBasicOps(); }
ReduceByKey extends StateAwareWindowWiseSingleInputOperator<
        IN, IN, IN, KEY, Pair<KEY, OUT>, W,
        ReduceByKey<IN, KEY, VALUE, OUT, W>> { public static <IN> KeyByBuilder<IN> of(Dataset<IN> input) { return new KeyByBuilder<>("ReduceByKey", input); } @SuppressWarnings("unchecked")  ReduceByKey(String name,
              Flow flow,
              Dataset<IN> input,
              UnaryFunction<IN, KEY> keyExtractor,
              @Nullable Class<KEY> keyClass,
              UnaryFunction<IN, VALUE> valueExtractor,
              @Nullable Windowing<IN, W> windowing,
              CombinableReduceFunction<OUT> reducer,
              Set<OutputHint> outputHints);   ReduceByKey(String name,
              Flow flow,
              Dataset<IN> input,
              UnaryFunction<IN, KEY> keyExtractor,
              @Nullable Class<KEY> keyClass,
              UnaryFunction<IN, VALUE> valueExtractor,
              @Nullable Windowing<IN, W> windowing,
              ReduceFunctor<VALUE, OUT> reducer,
              @Nullable BinaryFunction<VALUE, VALUE, Integer> valueComparator,
              Set<OutputHint> outputHints); static KeyByBuilder<IN> of(Dataset<IN> input); static OfBuilder named(String name); @Nullable Class<KEY> getKeyClass(); ReduceFunctor<VALUE, OUT> getReducer(); boolean isCombinable(); UnaryFunction<IN, VALUE> getValueExtractor(); @Nullable BinaryFunction<VALUE, VALUE, Integer> getValueComparator(); @SuppressWarnings("unchecked") @Override DAG<Operator<?, ?>> getBasicOps();  }