SumByKey extends ShuffleOperator<InputT, KeyT, KV<KeyT, Long>> implements CompositeOperator<InputT, KV<KeyT, Long>> { public static <InputT> KeyByBuilder<InputT> of(PCollection<InputT> input) { return named(null).of(input); } private  SumByKey(
      @Nullable String name,
      UnaryFunction<InputT, KeyT> keyExtractor,
      @Nullable TypeDescriptor<KeyT> keyType,
      UnaryFunction<InputT, Long> valueExtractor,
      @Nullable Window<InputT> window,
      TypeDescriptor<KV<KeyT, Long>> outputType); static KeyByBuilder<InputT> of(PCollection<InputT> input); static OfBuilder named(@Nullable String name); UnaryFunction<InputT, Long> getValueExtractor(); @Override PCollection<KV<KeyT, Long>> expand(PCollectionList<InputT> inputs);  }
@Test public void testBuild_ImplicitName() { final PCollection<String> dataset = TestUtils.createMockDataset(TypeDescriptors.strings()); final PCollection<KV<String, Long>> counted = SumByKey.of(dataset).keyBy(s -> s).output(); final SumByKey sum = (SumByKey) TestUtils.getProducer(counted); assertFalse(sum.getName().isPresent()); }
@Test public void testBuild_Windowing() { final PCollection<String> dataset = TestUtils.createMockDataset(TypeDescriptors.strings()); final PCollection<KV<String, Long>> counted = SumByKey.of(dataset) .keyBy(s -> s) .valueBy(s -> 1L) .windowBy(FixedWindows.of(org.joda.time.Duration.standardHours(1))) .triggeredBy(DefaultTrigger.of()) .discardingFiredPanes() .withAllowedLateness(Duration.millis(1000)) .output(); final SumByKey sum = (SumByKey) TestUtils.getProducer(counted); assertTrue(sum.getWindow().isPresent()); @SuppressWarnings("unchecked") final WindowDesc<?> windowDesc = WindowDesc.of((Window) sum.getWindow().get()); assertEquals( FixedWindows.of(org.joda.time.Duration.standardHours(1)), windowDesc.getWindowFn()); assertEquals(DefaultTrigger.of(), windowDesc.getTrigger()); assertEquals(AccumulationMode.DISCARDING_FIRED_PANES, windowDesc.getAccumulationMode()); assertEquals(Duration.millis(1000), windowDesc.getAllowedLateness()); }
@Test public void testWindow_applyIf() { final PCollection<String> dataset = TestUtils.createMockDataset(TypeDescriptors.strings()); final PCollection<KV<String, Long>> counted = SumByKey.of(dataset) .keyBy(s -> s) .valueBy(s -> 1L) .applyIf( true, b -> b.windowBy(FixedWindows.of(org.joda.time.Duration.standardHours(1))) .triggeredBy(DefaultTrigger.of()) .accumulationMode(AccumulationMode.DISCARDING_FIRED_PANES)) .output(); final SumByKey sum = (SumByKey) TestUtils.getProducer(counted); assertTrue(sum.getWindow().isPresent()); final Window<?> window = (Window) sum.getWindow().get(); assertEquals(FixedWindows.of(org.joda.time.Duration.standardHours(1)), window.getWindowFn()); assertEquals(DefaultTrigger.of(), WindowDesc.of(window).getTrigger()); assertEquals( AccumulationMode.DISCARDING_FIRED_PANES, WindowDesc.of(window).getAccumulationMode()); }