You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kylin.apache.org by li...@apache.org on 2015/08/26 10:08:24 UTC
incubator-kylin git commit: KYLIN-943 Add DoubleDeltaSerializer that
compactly serializes topn counters
Repository: incubator-kylin
Updated Branches:
refs/heads/0.8 11748806c -> 0b87b9a4a
KYLIN-943 Add DoubleDeltaSerializer that compactly serializes topn counters
Project: http://git-wip-us.apache.org/repos/asf/incubator-kylin/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-kylin/commit/0b87b9a4
Tree: http://git-wip-us.apache.org/repos/asf/incubator-kylin/tree/0b87b9a4
Diff: http://git-wip-us.apache.org/repos/asf/incubator-kylin/diff/0b87b9a4
Branch: refs/heads/0.8
Commit: 0b87b9a4a0b8ac0245860cadfce4e736f478d421
Parents: 1174880
Author: Li, Yang <ya...@ebay.com>
Authored: Wed Aug 26 16:07:37 2015 +0800
Committer: Li, Yang <ya...@ebay.com>
Committed: Wed Aug 26 16:08:12 2015 +0800
----------------------------------------------------------------------
.../common/topn/DoubleDeltaSerializer.java | 187 ++++++++++++++
.../common/hll/HyperLogLogCounterTest.java | 245 +++++++++++++++++++
.../common/topn/DoubleDeltaSerializerTest.java | 85 +++++++
.../common/util/HyperLogLogCounterTest.java | 244 ------------------
4 files changed, 517 insertions(+), 244 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/0b87b9a4/core-common/src/main/java/org/apache/kylin/common/topn/DoubleDeltaSerializer.java
----------------------------------------------------------------------
diff --git a/core-common/src/main/java/org/apache/kylin/common/topn/DoubleDeltaSerializer.java b/core-common/src/main/java/org/apache/kylin/common/topn/DoubleDeltaSerializer.java
new file mode 100644
index 0000000..cd798e0
--- /dev/null
+++ b/core-common/src/main/java/org/apache/kylin/common/topn/DoubleDeltaSerializer.java
@@ -0,0 +1,187 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kylin.common.topn;
+
+import java.nio.ByteBuffer;
+
+/**
+ * Given an array of double in ascending order, serialize them into very compact bytes.
+ *
+ * http://bitcharmer.blogspot.co.uk/2013/12/how-to-serialise-array-of-doubles-with.html
+ */
+public class DoubleDeltaSerializer {
+
+ // first 32 bits stores meta info
+ static final int PRECISION_BITS = 3;
+ static final int DELTA_SIZE_BITS = 6;
+ static final int LENGTH_BITS = 23;
+
+ static final long[] MASKS = new long[64];
+ {
+ for (int i = 0; i < MASKS.length; i++) {
+ MASKS[i] = (1 << i) - 1;
+ }
+ }
+
+ final private int precision;
+ final private int multiplier;
+
+ transient long[] reuseDeltas;
+
+ public DoubleDeltaSerializer() {
+ this(2);
+ }
+
+ /**
+ * @param precision round to specified number of digits after decimal point
+ */
+ public DoubleDeltaSerializer(int precision) {
+ checkFitInBits(precision, PRECISION_BITS);
+
+ this.precision = precision;
+ this.multiplier = (int) Math.pow(10, precision);
+ }
+
+ public void serialize(double values[], ByteBuffer buf) {
+ long[] deltas = calculateDeltas(values);
+ int deltaSize = maxDeltaSize(deltas, values.length - 1);
+
+ checkFitInBits(deltaSize, DELTA_SIZE_BITS);
+ checkFitInBits(values.length, LENGTH_BITS);
+
+ int meta = (int) precision << (LENGTH_BITS + DELTA_SIZE_BITS) | (int) deltaSize << LENGTH_BITS | values.length;
+ buf.putInt(meta);
+
+ if (values.length == 0)
+ return;
+
+ buf.putLong(roundAndPromote(values[0]));
+ putDeltas(deltas, values.length - 1, deltaSize, buf);
+ }
+
+ private void putDeltas(long[] deltas, int len, int deltaSize, ByteBuffer buf) {
+ long bits = 0;
+ int free = Long.SIZE;
+ for (int i = 0; i < len; i++) {
+ if (free >= deltaSize) {
+ bits |= deltas[i] << (free - deltaSize);
+ free -= deltaSize;
+ if (free == 0) {
+ buf.putLong(bits);
+ bits = 0;
+ free = Long.SIZE;
+ }
+ } else {
+ bits |= deltas[i] >> (deltaSize - free);
+ buf.putLong(bits);
+ free = (Long.SIZE - (deltaSize - free));
+ bits = deltas[i] << free;
+ }
+ }
+
+ if (free < Long.SIZE) {
+ buf.putLong(bits);
+ }
+ }
+
+ private int maxDeltaSize(long[] deltas, int len) {
+ long maxDelta = 0;
+ for (int i = 0; i < len; i++) {
+ maxDelta = Math.max(maxDelta, deltas[i]);
+ }
+ return Long.SIZE - Long.numberOfLeadingZeros(maxDelta);
+ }
+
+ private long[] calculateDeltas(double[] values) {
+ int len = values.length - 1;
+ len = Math.max(0, len);
+ if (reuseDeltas == null || reuseDeltas.length < len) {
+ reuseDeltas = new long[len];
+ }
+
+ if (len == 0)
+ return reuseDeltas;
+
+ long current = roundAndPromote(values[0]);
+ for (int i = 0; i < len; i++) {
+ long next = roundAndPromote(values[i + 1]);
+ reuseDeltas[i] = next - current;
+ assert reuseDeltas[i] >= 0;
+ current = next;
+ }
+ return reuseDeltas;
+ }
+
+ private long roundAndPromote(double value) {
+ return (long) (value * multiplier + .5d);
+ }
+
+ private void checkFitInBits(int num, int bits) {
+ if (num >= (1 << bits))
+ throw new IllegalArgumentException();
+ }
+
+ public double[] deserialize(ByteBuffer buf) {
+ int meta = buf.getInt();
+ int len = meta & ((1 << LENGTH_BITS) - 1);
+
+ double[] result = new double[len];
+ deserialize(buf, meta, result);
+ return result;
+ }
+
+ public int deserialize(ByteBuffer buf, double[] result) {
+ return deserialize(buf, buf.getInt(), result);
+ }
+
+ private int deserialize(ByteBuffer buf, int meta, double[] result) {
+ int precision = meta >>> (DELTA_SIZE_BITS + LENGTH_BITS);
+ assert precision == this.precision;
+ int deltaSize = (meta >>> LENGTH_BITS) & ((1 << DELTA_SIZE_BITS) - 1);
+ int len = meta & ((1 << LENGTH_BITS) - 1);
+
+ if (len == 0)
+ return 0;
+ assert result.length >= len;
+
+ long current = buf.getLong();
+ result[0] = (double) current / multiplier;
+
+ long bits = 0;
+ int left = 0;
+ for (int i = 1; i < len; i++) {
+ long delta = 0;
+
+ if (left >= deltaSize) {
+ delta = (bits >> (left - deltaSize)) & MASKS[deltaSize];
+ left -= deltaSize;
+ } else {
+ int more = deltaSize - left;
+ delta = (bits & MASKS[left]) << more;
+ bits = buf.getLong();
+ left = Long.SIZE;
+ delta |= (bits >> (left - more)) & MASKS[more];
+ left -= more;
+ }
+
+ current += delta;
+ result[i] = (double) current / multiplier;
+ }
+ return len;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/0b87b9a4/core-common/src/test/java/org/apache/kylin/common/hll/HyperLogLogCounterTest.java
----------------------------------------------------------------------
diff --git a/core-common/src/test/java/org/apache/kylin/common/hll/HyperLogLogCounterTest.java b/core-common/src/test/java/org/apache/kylin/common/hll/HyperLogLogCounterTest.java
new file mode 100644
index 0000000..0b008bc
--- /dev/null
+++ b/core-common/src/test/java/org/apache/kylin/common/hll/HyperLogLogCounterTest.java
@@ -0,0 +1,245 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+*/
+
+package org.apache.kylin.common.hll;
+
+import static org.junit.Assert.assertEquals;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.HashSet;
+import java.util.Random;
+import java.util.Set;
+
+import org.apache.kylin.common.hll.HyperLogLogPlusCounter;
+import org.apache.kylin.common.util.Bytes;
+import org.junit.Assert;
+import org.junit.Test;
+
+/**
+ * @author yangli9
+ *
+ */
+public class HyperLogLogCounterTest {
+
+ ByteBuffer buf = ByteBuffer.allocate(1024 * 1024);
+ Random rand1 = new Random(1);
+ Random rand2 = new Random(2);
+ Random rand3 = new Random(3);
+ int errorCount1 = 0;
+ int errorCount2 = 0;
+ int errorCount3 = 0;
+
+ @Test
+ public void testPeekLength() throws IOException {
+ HyperLogLogPlusCounter hllc = new HyperLogLogPlusCounter(10);
+ HyperLogLogPlusCounter copy = new HyperLogLogPlusCounter(10);
+ byte[] value = new byte[10];
+ for (int i = 0; i < 200000; i++) {
+ rand1.nextBytes(value);
+ hllc.add(value);
+
+ buf.clear();
+ hllc.writeRegisters(buf);
+
+ int len = buf.position();
+ buf.position(0);
+ assertEquals(len, hllc.peekLength(buf));
+
+ copy.readRegisters(buf);
+ assertEquals(len, buf.position());
+ assertEquals(hllc, copy);
+ }
+ buf.clear();
+ }
+
+ private Set<String> generateTestData(int n) {
+ Set<String> testData = new HashSet<String>();
+ for (int i = 0; i < n; i++) {
+ String[] samples = generateSampleData();
+ for (String sample : samples) {
+ testData.add(sample);
+ }
+ }
+ return testData;
+ }
+
+ // simulate the visit (=visitor+id)
+ private String[] generateSampleData() {
+
+ StringBuilder buf = new StringBuilder();
+ for (int i = 0; i < 19; i++) {
+ buf.append(Math.abs(rand1.nextInt()) % 10);
+ }
+ String header = buf.toString();
+
+ int size = Math.abs(rand3.nextInt()) % 9 + 1;
+ String[] samples = new String[size];
+ for (int k = 0; k < size; k++) {
+ buf = new StringBuilder(header);
+ buf.append("-");
+ for (int i = 0; i < 10; i++) {
+ buf.append(Math.abs(rand3.nextInt()) % 10);
+ }
+ samples[k] = buf.toString();
+ }
+
+ return samples;
+ }
+
+ @Test
+ public void countTest() throws IOException {
+ int n = 10;
+ for (int i = 0; i < 5; i++) {
+ count(n);
+ n *= 10;
+ }
+ }
+
+ private void count(int n) throws IOException {
+ Set<String> testSet = generateTestData(n);
+
+ HyperLogLogPlusCounter hllc = newHLLC();
+ for (String testData : testSet) {
+ hllc.add(Bytes.toBytes(testData));
+ }
+ long estimate = hllc.getCountEstimate();
+ double errorRate = hllc.getErrorRate();
+ double actualError = (double) Math.abs(testSet.size() - estimate) / testSet.size();
+ System.out.println(estimate);
+ System.out.println(testSet.size());
+ System.out.println(errorRate);
+ System.out.println("=" + actualError);
+ Assert.assertTrue(actualError < errorRate * 3.0);
+
+ checkSerialize(hllc);
+ }
+
+ private void checkSerialize(HyperLogLogPlusCounter hllc) throws IOException {
+ long estimate = hllc.getCountEstimate();
+ buf.clear();
+ hllc.writeRegisters(buf);
+ buf.flip();
+ hllc.readRegisters(buf);
+ Assert.assertEquals(estimate, hllc.getCountEstimate());
+ }
+
+ @Test
+ public void mergeTest() throws IOException {
+ double error = 0;
+ double absError = 0;
+ int n = 100;
+ for (int i = 0; i < n; i++) {
+ double e = merge();
+ error += e;
+ absError += Math.abs(e);
+ }
+ System.out.println("Total average error is " + error / n + " and absolute error is " + absError / n);
+
+ System.out.println(" errorRateCount1 is " + errorCount1 + "!");
+ System.out.println(" errorRateCount2 is " + errorCount2 + "!");
+ System.out.println(" errorRateCount3 is " + errorCount3 + "!");
+
+ Assert.assertTrue(errorCount1 <= n * 0.40);
+ Assert.assertTrue(errorCount2 <= n * 0.08);
+ Assert.assertTrue(errorCount3 <= n * 0.02);
+ }
+
+ private double merge() throws IOException {
+
+ int ln = 50;
+ int dn = 300;
+ Set<String> testSet = new HashSet<String>();
+ HyperLogLogPlusCounter[] hllcs = new HyperLogLogPlusCounter[ln];
+ for (int i = 0; i < ln; i++) {
+ hllcs[i] = newHLLC();
+ for (int k = 0; k < dn; k++) {
+ String[] samples = generateSampleData();
+ for (String data : samples) {
+ testSet.add(data);
+ hllcs[i].add(Bytes.toBytes(data));
+ }
+ }
+ }
+ HyperLogLogPlusCounter mergeHllc = newHLLC();
+ for (HyperLogLogPlusCounter hllc : hllcs) {
+ mergeHllc.merge(hllc);
+ checkSerialize(mergeHllc);
+ }
+
+ double errorRate = mergeHllc.getErrorRate();
+ long estimate = mergeHllc.getCountEstimate();
+ double actualError = (double) (testSet.size() - estimate) / testSet.size();
+
+ System.out.println(testSet.size() + "-" + estimate + " ~ " + actualError);
+
+ if (Math.abs(actualError) > errorRate) {
+ errorCount1++;
+ }
+ if (Math.abs(actualError) > 2 * errorRate) {
+ errorCount2++;
+ }
+ if (Math.abs(actualError) > 3 * errorRate) {
+ errorCount3++;
+ }
+
+ return actualError;
+ }
+
+ @Test
+ public void testPerformance() throws IOException {
+ int N = 3; // reduce N HLLC into one
+ int M = 1000; // for M times, use 100000 for real perf test
+
+ HyperLogLogPlusCounter samples[] = new HyperLogLogPlusCounter[N];
+ for (int i = 0; i < N; i++) {
+ samples[i] = newHLLC();
+ for (String str : generateTestData(10000))
+ samples[i].add(str);
+ }
+
+ System.out.println("Perf test running ... ");
+ long start = System.currentTimeMillis();
+ HyperLogLogPlusCounter sum = newHLLC();
+ for (int i = 0; i < M; i++) {
+ sum.clear();
+ for (int j = 0; j < N; j++) {
+ sum.merge(samples[j]);
+ checkSerialize(sum);
+ }
+ }
+ long duration = System.currentTimeMillis() - start;
+ System.out.println("Perf test result: " + duration / 1000 + " seconds");
+ }
+
+ @Test
+ public void testEquivalence() {
+ byte[] a = new byte[] { 0, 3, 4, 42, 2, 2 };
+ byte[] b = new byte[] { 3, 4, 42 };
+ HyperLogLogPlusCounter ha = new HyperLogLogPlusCounter();
+ HyperLogLogPlusCounter hb = new HyperLogLogPlusCounter();
+ ha.add(a, 1, 3);
+ hb.add(b);
+
+ Assert.assertTrue(ha.getCountEstimate() == hb.getCountEstimate());
+ }
+
+ private HyperLogLogPlusCounter newHLLC() {
+ return new HyperLogLogPlusCounter(16);
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/0b87b9a4/core-common/src/test/java/org/apache/kylin/common/topn/DoubleDeltaSerializerTest.java
----------------------------------------------------------------------
diff --git a/core-common/src/test/java/org/apache/kylin/common/topn/DoubleDeltaSerializerTest.java b/core-common/src/test/java/org/apache/kylin/common/topn/DoubleDeltaSerializerTest.java
new file mode 100644
index 0000000..7df26a8
--- /dev/null
+++ b/core-common/src/test/java/org/apache/kylin/common/topn/DoubleDeltaSerializerTest.java
@@ -0,0 +1,85 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kylin.common.topn;
+
+import static org.junit.Assert.*;
+
+import java.nio.ByteBuffer;
+import java.util.Arrays;
+import java.util.Random;
+
+import org.junit.Test;
+
+public class DoubleDeltaSerializerTest {
+
+ ByteBuffer buf = ByteBuffer.allocate(8192);
+ DoubleDeltaSerializer dds = new DoubleDeltaSerializer();
+
+ @Test
+ public void testEmpty() {
+ buf.clear();
+ dds.serialize(new double[] {}, buf);
+ buf.flip();
+ double[] r = dds.deserialize(buf);
+ assertTrue(r.length == 0);
+ }
+
+ @Test
+ public void testSingle() {
+ buf.clear();
+ dds.serialize(new double[] { 1.2 }, buf);
+ buf.flip();
+ double[] r = dds.deserialize(buf);
+ assertArrayEquals(new double[] { 1.2 }, r);
+ }
+
+ @Test
+ public void testRounding() {
+ buf.clear();
+ dds.serialize(new double[] { 1.234, 2.345 }, buf);
+ buf.flip();
+ double[] r = dds.deserialize(buf);
+ assertArrayEquals(new double[] { 1.23, 2.35 }, r);
+ }
+
+ @Test
+ public void testRandom() {
+ Random rand = new Random();
+ int n = 1000;
+
+ double[] nums = new double[n];
+ for (int i = 0; i < n; i++) {
+ nums[i] = rand.nextDouble() * 1000000;
+ }
+ Arrays.sort(nums);
+
+ buf.clear();
+ dds.serialize(nums, buf);
+ buf.flip();
+ double[] r = dds.deserialize(buf);
+ assertArrayEquals(nums, r);
+ System.out.println("doubles size of " + (n * 8) + " bytes serialized to " + buf.limit() + " bytes");
+ }
+
+ private static void assertArrayEquals(double[] expected, double[] actual) {
+ assertEquals(expected.length, actual.length);
+ for (int i = 0; i < expected.length; i++) {
+ assertEquals(expected[i], actual[i], 0.01);
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-kylin/blob/0b87b9a4/core-common/src/test/java/org/apache/kylin/common/util/HyperLogLogCounterTest.java
----------------------------------------------------------------------
diff --git a/core-common/src/test/java/org/apache/kylin/common/util/HyperLogLogCounterTest.java b/core-common/src/test/java/org/apache/kylin/common/util/HyperLogLogCounterTest.java
deleted file mode 100644
index 1485c66..0000000
--- a/core-common/src/test/java/org/apache/kylin/common/util/HyperLogLogCounterTest.java
+++ /dev/null
@@ -1,244 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
-*/
-
-package org.apache.kylin.common.util;
-
-import static org.junit.Assert.assertEquals;
-
-import java.io.IOException;
-import java.nio.ByteBuffer;
-import java.util.HashSet;
-import java.util.Random;
-import java.util.Set;
-
-import org.apache.kylin.common.hll.HyperLogLogPlusCounter;
-import org.junit.Assert;
-import org.junit.Test;
-
-/**
- * @author yangli9
- *
- */
-public class HyperLogLogCounterTest {
-
- ByteBuffer buf = ByteBuffer.allocate(1024 * 1024);
- Random rand1 = new Random(1);
- Random rand2 = new Random(2);
- Random rand3 = new Random(3);
- int errorCount1 = 0;
- int errorCount2 = 0;
- int errorCount3 = 0;
-
- @Test
- public void testPeekLength() throws IOException {
- HyperLogLogPlusCounter hllc = new HyperLogLogPlusCounter(10);
- HyperLogLogPlusCounter copy = new HyperLogLogPlusCounter(10);
- byte[] value = new byte[10];
- for (int i = 0; i < 200000; i++) {
- rand1.nextBytes(value);
- hllc.add(value);
-
- buf.clear();
- hllc.writeRegisters(buf);
-
- int len = buf.position();
- buf.position(0);
- assertEquals(len, hllc.peekLength(buf));
-
- copy.readRegisters(buf);
- assertEquals(len, buf.position());
- assertEquals(hllc, copy);
- }
- buf.clear();
- }
-
- private Set<String> generateTestData(int n) {
- Set<String> testData = new HashSet<String>();
- for (int i = 0; i < n; i++) {
- String[] samples = generateSampleData();
- for (String sample : samples) {
- testData.add(sample);
- }
- }
- return testData;
- }
-
- // simulate the visit (=visitor+id)
- private String[] generateSampleData() {
-
- StringBuilder buf = new StringBuilder();
- for (int i = 0; i < 19; i++) {
- buf.append(Math.abs(rand1.nextInt()) % 10);
- }
- String header = buf.toString();
-
- int size = Math.abs(rand3.nextInt()) % 9 + 1;
- String[] samples = new String[size];
- for (int k = 0; k < size; k++) {
- buf = new StringBuilder(header);
- buf.append("-");
- for (int i = 0; i < 10; i++) {
- buf.append(Math.abs(rand3.nextInt()) % 10);
- }
- samples[k] = buf.toString();
- }
-
- return samples;
- }
-
- @Test
- public void countTest() throws IOException {
- int n = 10;
- for (int i = 0; i < 5; i++) {
- count(n);
- n *= 10;
- }
- }
-
- private void count(int n) throws IOException {
- Set<String> testSet = generateTestData(n);
-
- HyperLogLogPlusCounter hllc = newHLLC();
- for (String testData : testSet) {
- hllc.add(Bytes.toBytes(testData));
- }
- long estimate = hllc.getCountEstimate();
- double errorRate = hllc.getErrorRate();
- double actualError = (double) Math.abs(testSet.size() - estimate) / testSet.size();
- System.out.println(estimate);
- System.out.println(testSet.size());
- System.out.println(errorRate);
- System.out.println("=" + actualError);
- Assert.assertTrue(actualError < errorRate * 3.0);
-
- checkSerialize(hllc);
- }
-
- private void checkSerialize(HyperLogLogPlusCounter hllc) throws IOException {
- long estimate = hllc.getCountEstimate();
- buf.clear();
- hllc.writeRegisters(buf);
- buf.flip();
- hllc.readRegisters(buf);
- Assert.assertEquals(estimate, hllc.getCountEstimate());
- }
-
- @Test
- public void mergeTest() throws IOException {
- double error = 0;
- double absError = 0;
- int n = 100;
- for (int i = 0; i < n; i++) {
- double e = merge();
- error += e;
- absError += Math.abs(e);
- }
- System.out.println("Total average error is " + error / n + " and absolute error is " + absError / n);
-
- System.out.println(" errorRateCount1 is " + errorCount1 + "!");
- System.out.println(" errorRateCount2 is " + errorCount2 + "!");
- System.out.println(" errorRateCount3 is " + errorCount3 + "!");
-
- Assert.assertTrue(errorCount1 <= n * 0.40);
- Assert.assertTrue(errorCount2 <= n * 0.08);
- Assert.assertTrue(errorCount3 <= n * 0.02);
- }
-
- private double merge() throws IOException {
-
- int ln = 50;
- int dn = 300;
- Set<String> testSet = new HashSet<String>();
- HyperLogLogPlusCounter[] hllcs = new HyperLogLogPlusCounter[ln];
- for (int i = 0; i < ln; i++) {
- hllcs[i] = newHLLC();
- for (int k = 0; k < dn; k++) {
- String[] samples = generateSampleData();
- for (String data : samples) {
- testSet.add(data);
- hllcs[i].add(Bytes.toBytes(data));
- }
- }
- }
- HyperLogLogPlusCounter mergeHllc = newHLLC();
- for (HyperLogLogPlusCounter hllc : hllcs) {
- mergeHllc.merge(hllc);
- checkSerialize(mergeHllc);
- }
-
- double errorRate = mergeHllc.getErrorRate();
- long estimate = mergeHllc.getCountEstimate();
- double actualError = (double) (testSet.size() - estimate) / testSet.size();
-
- System.out.println(testSet.size() + "-" + estimate + " ~ " + actualError);
-
- if (Math.abs(actualError) > errorRate) {
- errorCount1++;
- }
- if (Math.abs(actualError) > 2 * errorRate) {
- errorCount2++;
- }
- if (Math.abs(actualError) > 3 * errorRate) {
- errorCount3++;
- }
-
- return actualError;
- }
-
- @Test
- public void testPerformance() throws IOException {
- int N = 3; // reduce N HLLC into one
- int M = 1000; // for M times, use 100000 for real perf test
-
- HyperLogLogPlusCounter samples[] = new HyperLogLogPlusCounter[N];
- for (int i = 0; i < N; i++) {
- samples[i] = newHLLC();
- for (String str : generateTestData(10000))
- samples[i].add(str);
- }
-
- System.out.println("Perf test running ... ");
- long start = System.currentTimeMillis();
- HyperLogLogPlusCounter sum = newHLLC();
- for (int i = 0; i < M; i++) {
- sum.clear();
- for (int j = 0; j < N; j++) {
- sum.merge(samples[j]);
- checkSerialize(sum);
- }
- }
- long duration = System.currentTimeMillis() - start;
- System.out.println("Perf test result: " + duration / 1000 + " seconds");
- }
-
- @Test
- public void testEquivalence() {
- byte[] a = new byte[] { 0, 3, 4, 42, 2, 2 };
- byte[] b = new byte[] { 3, 4, 42 };
- HyperLogLogPlusCounter ha = new HyperLogLogPlusCounter();
- HyperLogLogPlusCounter hb = new HyperLogLogPlusCounter();
- ha.add(a, 1, 3);
- hb.add(b);
-
- Assert.assertTrue(ha.getCountEstimate() == hb.getCountEstimate());
- }
-
- private HyperLogLogPlusCounter newHLLC() {
- return new HyperLogLogPlusCounter(16);
- }
-}