Distinct extends ShuffleOperator<InputT, KeyT, InputT> implements CompositeOperator<InputT, InputT> { public static <InputT> ProjectedBuilder<InputT, InputT> of(PCollection<InputT> input) { return named(null).of(input); } private  Distinct(
      @Nullable String name,
      UnaryFunction<InputT, KeyT> transform,
      @Nullable TypeDescriptor<InputT> outputType,
      @Nullable TypeDescriptor<KeyT> projectedType,
      @Nullable Window<InputT> window,
      @Nullable SelectionPolicy policy,
      boolean projected); static ProjectedBuilder<InputT, InputT> of(PCollection<InputT> input); static OfBuilder named(@Nullable String name); @Override PCollection<InputT> expand(PCollectionList<InputT> inputs);  }
@Test public void testBuild_ImplicitName() { final PCollection<String> dataset = TestUtils.createMockDataset(TypeDescriptors.strings()); final PCollection<String> uniq = Distinct.of(dataset).output(); final Distinct distinct = (Distinct) TestUtils.getProducer(uniq); assertFalse(distinct.getName().isPresent()); }
@Test public void testBuild_Windowing() { final PCollection<String> dataset = TestUtils.createMockDataset(TypeDescriptors.strings()); final PCollection<String> uniq = Distinct.of(dataset) .windowBy(FixedWindows.of(org.joda.time.Duration.standardHours(1))) .triggeredBy(DefaultTrigger.of()) .accumulationMode(AccumulationMode.DISCARDING_FIRED_PANES) .output(); final Distinct distinct = (Distinct) TestUtils.getProducer(uniq); assertTrue(distinct.getWindow().isPresent()); @SuppressWarnings("unchecked") final WindowDesc<?> windowDesc = WindowDesc.of((Window) distinct.getWindow().get()); assertEquals( FixedWindows.of(org.joda.time.Duration.standardHours(1)), windowDesc.getWindowFn()); assertEquals(DefaultTrigger.of(), windowDesc.getTrigger()); }
@Test public void testWindow_applyIf() { final PCollection<String> dataset = TestUtils.createMockDataset(TypeDescriptors.strings()); final PCollection<String> uniq = Distinct.of(dataset) .applyIf( true, b -> b.windowBy(FixedWindows.of(Duration.standardHours(1))) .triggeredBy(DefaultTrigger.of()) .discardingFiredPanes()) .output(); final Distinct distinct = (Distinct) TestUtils.getProducer(uniq); assertTrue(distinct.getWindow().isPresent()); @SuppressWarnings("unchecked") final WindowDesc<?> windowDesc = WindowDesc.of((Window) distinct.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()); }