Deduplicate { public static <T, IdT> Deduplicate.WithRepresentativeValues<T, IdT> withRepresentativeValueFn( SerializableFunction<T, IdT> representativeValueFn) { return new Deduplicate.WithRepresentativeValues<T, IdT>( DEFAULT_TIME_DOMAIN, DEFAULT_DURATION, representativeValueFn, null, null); } 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, UsesTestStreamWithProcessingTime.class}) public void testRepresentativeValuesWithCoder() { Instant base = new Instant(0); TestStream<KV<Integer, String>> values = TestStream.create(KvCoder.of(VarIntCoder.of(), StringUtf8Coder.of())) .advanceWatermarkTo(base) .addElements( TimestampedValue.of(KV.of(1, "k1"), base), TimestampedValue.of(KV.of(2, "k2"), base.plus(Duration.standardSeconds(10))), TimestampedValue.of(KV.of(3, "k3"), base.plus(Duration.standardSeconds(20)))) .advanceProcessingTime(Duration.standardMinutes(1)) .addElements( TimestampedValue.of(KV.of(1, "k1"), base.plus(Duration.standardSeconds(30))), TimestampedValue.of(KV.of(2, "k2"), base.plus(Duration.standardSeconds(40))), TimestampedValue.of(KV.of(3, "k3"), base.plus(Duration.standardSeconds(50)))) .advanceWatermarkToInfinity(); PCollection<KV<Integer, String>> distinctValues = p.apply(values) .apply( Deduplicate.withRepresentativeValueFn(new Keys<Integer>()) .withRepresentativeCoder(VarIntCoder.of())); PAssert.that(distinctValues).containsInAnyOrder(KV.of(1, "k1"), KV.of(2, "k2"), KV.of(3, "k3")); p.run(); }
@Test @Category({NeedsRunner.class, UsesTestStreamWithProcessingTime.class}) public void testTriggeredRepresentativeValuesWithType() { Instant base = new Instant(0); TestStream<KV<Integer, String>> values = TestStream.create(KvCoder.of(VarIntCoder.of(), StringUtf8Coder.of())) .advanceWatermarkTo(base) .addElements( TimestampedValue.of(KV.of(1, "k1"), base), TimestampedValue.of(KV.of(2, "k2"), base.plus(Duration.standardSeconds(10))), TimestampedValue.of(KV.of(3, "k3"), base.plus(Duration.standardSeconds(20)))) .advanceProcessingTime(Duration.standardMinutes(1)) .addElements( TimestampedValue.of(KV.of(1, "k1"), base.plus(Duration.standardSeconds(30))), TimestampedValue.of(KV.of(2, "k2"), base.plus(Duration.standardSeconds(40))), TimestampedValue.of(KV.of(3, "k3"), base.plus(Duration.standardSeconds(50)))) .advanceWatermarkToInfinity(); PCollection<KV<Integer, String>> distinctValues = p.apply(values) .apply( Deduplicate.withRepresentativeValueFn(new Keys<Integer>()) .withRepresentativeCoder(VarIntCoder.of())); PAssert.that(distinctValues).containsInAnyOrder(KV.of(1, "k1"), KV.of(2, "k2"), KV.of(3, "k3")); p.run(); }
@Test public void withLambdaRepresentativeValuesFnNoTypeDescriptorShouldThrow() { Multimap<Integer, String> predupedContents = HashMultimap.create(); predupedContents.put(3, "foo"); predupedContents.put(4, "foos"); predupedContents.put(6, "barbaz"); predupedContents.put(6, "bazbar"); PCollection<String> dupes = p.apply(Create.of("foo", "foos", "barbaz", "barbaz", "bazbar", "foo")); assertThrows( "Unable to return a default Coder for RemoveRepresentativeDupes", IllegalStateException.class, () -> dupes.apply( "RemoveRepresentativeDupes", Deduplicate.withRepresentativeValueFn(String::length))); }