ApproximateDistinct { public static <InputT> GloballyDistinct<InputT> globally() { return GloballyDistinct.<InputT>builder().build(); }  static GloballyDistinct<InputT> globally(); static PerKeyDistinct<K, V> perKey(); static long precisionForRelativeError(double relativeError); static double relativeErrorForPrecision(int p);  }
@Test public void smallCardinality() { final int smallCard = 1000; final int p = 6; final double expectedErr = 1.104 / Math.sqrt(p); List<Integer> small = new ArrayList<>(); for (int i = 0; i < smallCard; i++) { small.add(i); } PCollection<Long> cardinality = tp.apply("small stream", Create.of(small)) .apply("small cardinality", ApproximateDistinct.<Integer>globally().withPrecision(p)); PAssert.that("Not Accurate Enough", cardinality) .satisfies(new VerifyAccuracy(smallCard, expectedErr)); tp.run(); }
@Test public void bigCardinality() { final int cardinality = 15000; final int p = 15; final int sp = 20; 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> res = tp.apply("big stream", Create.of(stream)) .apply( "big cardinality", ApproximateDistinct.<Integer>globally().withPrecision(p).withSparsePrecision(sp)); PAssert.that("Verify Accuracy for big cardinality", res) .satisfies(new VerifyAccuracy(cardinality, expectedErr)); tp.run(); }
@Test public void customObject() { final int cardinality = 500; final int p = 15; final double expectedErr = 1.04 / Math.sqrt(p); Schema schema = SchemaBuilder.record("User") .fields() .requiredString("Pseudo") .requiredInt("Age") .endRecord(); List<GenericRecord> users = new ArrayList<>(); for (int i = 1; i <= cardinality; i++) { GenericData.Record newRecord = new GenericData.Record(schema); newRecord.put("Pseudo", "User" + i); newRecord.put("Age", i); users.add(newRecord); } PCollection<Long> results = tp.apply("Create stream", Create.of(users).withCoder(AvroCoder.of(schema))) .apply( "Test custom object", ApproximateDistinct.<GenericRecord>globally().withPrecision(p)); PAssert.that("Verify Accuracy for custom object", results) .satisfies(new VerifyAccuracy(cardinality, expectedErr)); tp.run(); }