TopPerKey extends ShuffleOperator<InputT, KeyT, Triple<KeyT, ValueT, ScoreT>> implements TypeAware.Value<ValueT>, CompositeOperator<InputT, Triple<KeyT, ValueT, ScoreT>> { public static <InputT> KeyByBuilder<InputT> of(PCollection<InputT> input) { return named(null).of(input); } private  TopPerKey(
      @Nullable String name,
      UnaryFunction<InputT, KeyT> keyExtractor,
      @Nullable TypeDescriptor<KeyT> keyType,
      UnaryFunction<InputT, ValueT> valueExtractor,
      @Nullable TypeDescriptor<ValueT> valueType,
      UnaryFunction<InputT, ScoreT> scoreExtractor,
      @Nullable TypeDescriptor<ScoreT> scoreType,
      @Nullable Window<InputT> window,
      @Nullable TypeDescriptor<Triple<KeyT, ValueT, ScoreT>> outputType); static KeyByBuilder<InputT> of(PCollection<InputT> input); static OfBuilder named(@Nullable String name); UnaryFunction<InputT, ValueT> getValueExtractor(); @Override Optional<TypeDescriptor<ValueT>> getValueType(); UnaryFunction<InputT, ScoreT> getScoreExtractor(); Optional<TypeDescriptor<ScoreT>> getScoreType(); @Override PCollection<Triple<KeyT, ValueT, ScoreT>> expand(PCollectionList<InputT> inputs);  }
@Test public void testBuild_ImplicitName() { final PCollection<String> dataset = TestUtils.createMockDataset(TypeDescriptors.strings()); final PCollection<Triple<String, Long, Long>> result = TopPerKey.of(dataset).keyBy(s -> s).valueBy(s -> 1L).scoreBy(s -> 1L).output(); final TopPerKey tpk = (TopPerKey) TestUtils.getProducer(result); assertFalse(tpk.getName().isPresent()); }
@Test public void testBuild_Windowing() { final PCollection<String> dataset = TestUtils.createMockDataset(TypeDescriptors.strings()); final PCollection<Triple<String, Long, Long>> result = TopPerKey.of(dataset) .keyBy(s -> s) .valueBy(s -> 1L) .scoreBy(s -> 1L) .windowBy(FixedWindows.of(org.joda.time.Duration.standardHours(1))) .triggeredBy(DefaultTrigger.of()) .accumulationMode(AccumulationMode.DISCARDING_FIRED_PANES) .output(); final TopPerKey tpk = (TopPerKey) TestUtils.getProducer(result); assertTrue(tpk.getWindow().isPresent()); @SuppressWarnings("unchecked") final WindowDesc<?> windowDesc = WindowDesc.of((Window) tpk.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()); }
@Test public void testWindow_applyIf() { final PCollection<String> dataset = TestUtils.createMockDataset(TypeDescriptors.strings()); final PCollection<Triple<String, Long, Long>> result = TopPerKey.of(dataset) .keyBy(s -> s) .valueBy(s -> 1L) .scoreBy(s -> 1L) .applyIf( true, b -> b.windowBy(FixedWindows.of(org.joda.time.Duration.standardHours(1))) .triggeredBy(DefaultTrigger.of()) .accumulatingFiredPanes()) .output(); final TopPerKey tpk = (TopPerKey) TestUtils.getProducer(result); assertTrue(tpk.getWindow().isPresent()); @SuppressWarnings("unchecked") final WindowDesc<?> windowDesc = WindowDesc.of((Window) tpk.getWindow().get()); assertEquals( FixedWindows.of(org.joda.time.Duration.standardHours(1)), windowDesc.getWindowFn()); assertEquals(DefaultTrigger.of(), windowDesc.getTrigger()); assertEquals(AccumulationMode.ACCUMULATING_FIRED_PANES, windowDesc.getAccumulationMode()); }