public static <K> PerKeyDigest<K> perKey() {
    return PerKeyDigest.<K>builder().build();
  }
@Test
  public void perKey() {
    PCollection<KV<Double, Double>> col =
        tp.apply(Create.of(stream))
            .apply(WithKeys.of(1))
            .apply(TDigestQuantiles.<Integer>perKey().withCompression(compression))
            .apply(Values.create())
            .apply(ParDo.of(new RetrieveQuantiles(quantiles)));

    PAssert.that("Verify Accuracy", col).satisfies(new VerifyAccuracy());

    tp.run();
  }