Distinct extends PTransform<PCollection<T>, PCollection<T>> { public static <T> Distinct<T> create() { return new Distinct<>(); }  static Distinct<T> create(); static WithRepresentativeValues<T, IdT> withRepresentativeValueFn(
      SerializableFunction<T, IdT> fn); @Override PCollection<T> expand(PCollection<T> in);  }
@Test @Category(NeedsRunner.class) public void testDistinct() { List<String> strings = Arrays.asList("k1", "k5", "k5", "k2", "k1", "k2", "k3"); PCollection<String> input = p.apply(Create.of(strings).withCoder(StringUtf8Coder.of())); PCollection<String> output = input.apply(Distinct.create()); PAssert.that(output).containsInAnyOrder("k1", "k5", "k2", "k3"); p.run(); }
@Test @Category(NeedsRunner.class) public void testDistinctEmpty() { List<String> strings = Arrays.asList(); PCollection<String> input = p.apply(Create.of(strings).withCoder(StringUtf8Coder.of())); PCollection<String> output = input.apply(Distinct.create()); PAssert.that(output).empty(); p.run(); }
@Test @Category({NeedsRunner.class, UsesTestStream.class}) public void testWindowedDistinct() { Instant base = new Instant(0); TestStream<String> values = TestStream.create(StringUtf8Coder.of()) .advanceWatermarkTo(base) .addElements( TimestampedValue.of("k1", base), TimestampedValue.of("k2", base.plus(Duration.standardSeconds(10))), TimestampedValue.of("k3", base.plus(Duration.standardSeconds(20))), TimestampedValue.of("k1", base.plus(Duration.standardSeconds(30))), TimestampedValue.of("k2", base.plus(Duration.standardSeconds(40))), TimestampedValue.of("k3", base.plus(Duration.standardSeconds(50))), TimestampedValue.of("k4", base.plus(Duration.standardSeconds(60))), TimestampedValue.of("k5", base.plus(Duration.standardSeconds(70))), TimestampedValue.of("k6", base.plus(Duration.standardSeconds(80)))) .advanceWatermarkToInfinity(); PCollection<String> distinctValues = windowedDistinctPipeline .apply(values) .apply(Window.into(FixedWindows.of(Duration.standardSeconds(30)))) .apply(Distinct.create()); PAssert.that(distinctValues) .inWindow(new IntervalWindow(base, base.plus(Duration.standardSeconds(30)))) .containsInAnyOrder("k1", "k2", "k3"); PAssert.that(distinctValues) .inWindow( new IntervalWindow( base.plus(Duration.standardSeconds(30)), base.plus(Duration.standardSeconds(60)))) .containsInAnyOrder("k1", "k2", "k3"); PAssert.that(distinctValues) .inWindow( new IntervalWindow( base.plus(Duration.standardSeconds(60)), base.plus(Duration.standardSeconds(90)))) .containsInAnyOrder("k4", "k5", "k6"); windowedDistinctPipeline.run(); }
@Test @Category({NeedsRunner.class, UsesTestStreamWithProcessingTime.class}) public void testTriggeredDistinct() { Instant base = new Instant(0); TestStream<String> values = TestStream.create(StringUtf8Coder.of()) .advanceWatermarkTo(base) .addElements( TimestampedValue.of("k1", base), TimestampedValue.of("k2", base.plus(Duration.standardSeconds(10))), TimestampedValue.of("k3", base.plus(Duration.standardSeconds(20)))) .advanceProcessingTime(Duration.standardMinutes(1)) .addElements( TimestampedValue.of("k1", base.plus(Duration.standardSeconds(30))), TimestampedValue.of("k2", base.plus(Duration.standardSeconds(40))), TimestampedValue.of("k3", base.plus(Duration.standardSeconds(50)))) .advanceWatermarkToInfinity(); PCollection<String> distinctValues = triggeredDistinctPipeline .apply(values) .apply( Window.<String>into(FixedWindows.of(Duration.standardMinutes(1))) .triggering( Repeatedly.forever( AfterProcessingTime.pastFirstElementInPane() .plusDelayOf(Duration.standardSeconds(30)))) .withAllowedLateness(Duration.ZERO) .accumulatingFiredPanes()) .apply(Distinct.create()); PAssert.that(distinctValues).containsInAnyOrder("k1", "k2", "k3"); triggeredDistinctPipeline.run(); }