ApproximateDistinct { public static <K, V> PerKeyDistinct<K, V> perKey() { return PerKeyDistinct.<K, V>builder().build(); }  static GloballyDistinct<InputT> globally(); static PerKeyDistinct<K, V> perKey(); static long precisionForRelativeError(double relativeError); static double relativeErrorForPrecision(int p);  }
@Test public void perKey() { final int cardinality = 1000; final int p = 15; final double expectedErr = 1.04 / Math.sqrt(p); List<Integer> stream = new ArrayList<>(); for (int i = 1; i <= cardinality; i++) { stream.addAll(Collections.nCopies(2, i)); } Collections.shuffle(stream); PCollection<Long> results = tp.apply("per key stream", Create.of(stream)) .apply("create keys", WithKeys.of(1)) .apply( "per key cardinality", ApproximateDistinct.<Integer, Integer>perKey().withPrecision(p)) .apply("extract values", Values.create()); PAssert.that("Verify Accuracy for cardinality per key", results) .satisfies(new VerifyAccuracy(cardinality, expectedErr)); tp.run(); }