Distinct extends PTransform<PCollection<T>, PCollection<T>> { public static <T, IdT> WithRepresentativeValues<T, IdT> withRepresentativeValueFn( SerializableFunction<T, IdT> fn) { return new WithRepresentativeValues<>(fn, null); }  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 testDistinctWithRepresentativeValue() { List<KV<String, String>> strings = Arrays.asList(KV.of("k1", "v1"), KV.of("k1", "v2"), KV.of("k2", "v1")); PCollection<KV<String, String>> input = p.apply(Create.of(strings)); PCollection<KV<String, String>> output = input.apply( Distinct.withRepresentativeValueFn(new Keys<String>()) .withRepresentativeType(TypeDescriptor.of(String.class))); PAssert.that(output).satisfies(new Checker()); p.run(); }