ReduceByKey extends ShuffleOperator<InputT, KeyT, KV<KeyT, OutputT>> implements TypeAware.Value<ValueT> { public static <InputT> KeyByBuilder<InputT> of(PCollection<InputT> input) { return named(null).of(input); } private  ReduceByKey(
      @Nullable String name,
      UnaryFunction<InputT, KeyT> keyExtractor,
      @Nullable TypeDescriptor<KeyT> keyType,
      UnaryFunction<InputT, ValueT> valueExtractor,
      @Nullable TypeDescriptor<ValueT> valueType,
      ReduceFunctor<ValueT, OutputT> reducer,
      @Nullable BinaryFunction<ValueT, ValueT, Integer> valueComparator,
      @Nullable Window<InputT> window,
      @Nullable TypeDescriptor<KV<KeyT, OutputT>> outputType); private  ReduceByKey(
      @Nullable String name,
      UnaryFunction<InputT, KeyT> keyExtractor,
      @Nullable TypeDescriptor<KeyT> keyType,
      UnaryFunction<InputT, ValueT> valueExtractor,
      @Nullable TypeDescriptor<ValueT> valueType,
      VoidFunction<AccT> accumulatorFactory,
      BinaryFunction<AccT, ValueT, AccT> accumulate,
      CombinableBinaryFunction<AccT> mergeAccumulators,
      UnaryFunction<AccT, OutputT> outputFn,
      TypeDescriptor<AccT> accumulatorType,
      @Nullable BinaryFunction<ValueT, ValueT, Integer> valueComparator,
      @Nullable Window<InputT> window,
      @Nullable TypeDescriptor<KV<KeyT, OutputT>> outputType); static KeyByBuilder<InputT> of(PCollection<InputT> input); static OfBuilder named(@Nullable String name); boolean isCombineFnStyle(); ReduceFunctor<ValueT, OutputT> getReducer(); VoidFunction<AccT> getAccumulatorFactory(); BinaryFunction<AccT, ValueT, AccT> getAccumulate(); CombinableBinaryFunction<AccT> getMergeAccumulators(); UnaryFunction<AccT, OutputT> getOutputFn(); TypeDescriptor<AccT> getAccumulatorType(); boolean isCombinable(); UnaryFunction<InputT, ValueT> getValueExtractor(); Optional<BinaryFunction<ValueT, ValueT, Integer>> getValueComparator(); @Override Optional<TypeDescriptor<ValueT>> getValueType();  }
@Test public void testBuild_ImplicitName() { final PCollection<String> dataset = TestUtils.createMockDataset(TypeDescriptors.strings()); final PCollection<KV<String, Long>> reduced = ReduceByKey.of(dataset).keyBy(s -> s).valueBy(s -> 1L).combineBy(Sums.ofLongs()).output(); final ReduceByKey reduce = (ReduceByKey) TestUtils.getProducer(reduced); assertFalse(reduce.getName().isPresent()); }
@Test public void testWindow_applyIf() { final PCollection<String> dataset = TestUtils.createMockDataset(TypeDescriptors.strings()); final PCollection<KV<String, Long>> reduced = ReduceByKey.of(dataset) .keyBy(s -> s) .valueBy(s -> 1L) .combineBy(Sums.ofLongs()) .applyIf( true, b -> b.windowBy(FixedWindows.of(Duration.standardHours(1))) .triggeredBy(DefaultTrigger.of()) .accumulationMode(AccumulationMode.DISCARDING_FIRED_PANES)) .output(); final ReduceByKey reduce = (ReduceByKey) TestUtils.getProducer(reduced); assertTrue(reduce.getWindow().isPresent()); @SuppressWarnings("unchecked") final Window<? extends BoundedWindow> window = (Window) reduce.getWindow().get(); assertEquals(FixedWindows.of(org.joda.time.Duration.standardHours(1)), window.getWindowFn()); assertEquals(DefaultTrigger.of(), WindowDesc.of(window).getTrigger()); assertSame( AccumulationMode.DISCARDING_FIRED_PANES, WindowDesc.of(window).getAccumulationMode()); }
@Test public void testWindow_applyIfNot() { final PCollection<String> dataset = TestUtils.createMockDataset(TypeDescriptors.strings()); final PCollection<KV<String, Long>> reduced = ReduceByKey.of(dataset) .keyBy(s -> s) .valueBy(s -> 1L) .combineBy(Sums.ofLongs()) .applyIf( false, b -> b.windowBy(FixedWindows.of(Duration.standardHours(1))) .triggeredBy(DefaultTrigger.of()) .accumulationMode(AccumulationMode.DISCARDING_FIRED_PANES)) .output(); final ReduceByKey reduce = (ReduceByKey) TestUtils.getProducer(reduced); assertFalse(reduce.getWindow().isPresent()); }
@Test @SuppressWarnings("unchecked") public void testTypeHints_typePropagation() { final PCollection<String> dataset = TestUtils.createMockDataset(TypeDescriptors.strings()); final TypeDescriptor<String> keyType = TypeDescriptors.strings(); final TypeDescriptor<Long> valueType = TypeDescriptors.longs(); final TypeDescriptor<Long> outputType = TypeDescriptors.longs(); final PCollection<KV<String, Long>> reduced = ReduceByKey.of(dataset) .keyBy(s -> s, keyType) .valueBy(s -> 1L, valueType) .combineBy(Sums.ofLongs()) .output(); final ReduceByKey reduce = (ReduceByKey) TestUtils.getProducer(reduced); TypePropagationAssert.assertOperatorTypeAwareness(reduce, keyType, valueType, outputType); }