public static <K, V> PerKeyDistinct<K, V> perKey() {
    return PerKeyDistinct.<K, V>builder().build();
  }
@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();
  }