Deduplicate { public static <T> Deduplicate.Values<T> values() { return new Deduplicate.Values<>(DEFAULT_TIME_DOMAIN, DEFAULT_DURATION); } private  Deduplicate(); static Deduplicate.Values<T> values(); static Deduplicate.KeyedValues<K, V> keyedValues(); static Deduplicate.WithRepresentativeValues<T, IdT> withRepresentativeValueFn(
      SerializableFunction<T, IdT> representativeValueFn); static final TimeDomain DEFAULT_TIME_DOMAIN; static final Duration DEFAULT_DURATION; }
@Test @Category({NeedsRunner.class, UsesTestStream.class}) public void testInDifferentWindows() { 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 = p.apply(values) .apply(Window.into(FixedWindows.of(Duration.standardSeconds(30)))) .apply(Deduplicate.values()); 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"); p.run(); }
@Test @Category({NeedsRunner.class, UsesTestStream.class}) public void testEventTime() { 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("maybedup", base.plus(Duration.standardSeconds(59)))) .advanceWatermarkTo(base.plus(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)))) .advanceWatermarkTo( base.plus(Duration.standardMinutes(1)).plus(Deduplicate.DEFAULT_DURATION)) .addElements(TimestampedValue.of("maybedup", base.plus(Duration.standardSeconds(59)))) .advanceWatermarkToInfinity(); PCollection<String> distinctValues = p.apply(values).apply(Deduplicate.<String>values().withTimeDomain(TimeDomain.EVENT_TIME)); PAssert.that(distinctValues) .satisfies( (Iterable<String> input) -> { assertEquals(1, Iterables.frequency(input, "k1")); assertEquals(1, Iterables.frequency(input, "k2")); assertEquals(1, Iterables.frequency(input, "k3")); assertTrue( Iterables.frequency(input, "maybedup") == 1 || Iterables.frequency(input, "maybedup") == 2); return null; }); p.run(); }
@Test @Category({NeedsRunner.class, UsesTestStreamWithProcessingTime.class}) public void testProcessingTime() { 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("maybedup", base.plus(Duration.standardSeconds(59)))) .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)))) .advanceProcessingTime(Deduplicate.DEFAULT_DURATION) .addElements(TimestampedValue.of("maybedup", base.plus(Duration.standardSeconds(59)))) .advanceWatermarkToInfinity(); PCollection<String> distinctValues = p.apply(values).apply(Deduplicate.values()); PAssert.that(distinctValues) .satisfies( (Iterable<String> input) -> { assertEquals(1, Iterables.frequency(input, "k1")); assertEquals(1, Iterables.frequency(input, "k2")); assertEquals(1, Iterables.frequency(input, "k3")); assertTrue( Iterables.frequency(input, "maybedup") == 1 || Iterables.frequency(input, "maybedup") == 2); return null; }); p.run(); }