You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@kylin.apache.org by GitBox <gi...@apache.org> on 2018/08/27 09:24:17 UTC
[GitHub] r7raul1984 closed pull request #211: KYLIN-3514 Support AUC UDAF
r7raul1984 closed pull request #211: KYLIN-3514 Support AUC UDAF
URL: https://github.com/apache/kylin/pull/211
This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:
As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):
diff --git a/core-metadata/pom.xml b/core-metadata/pom.xml
index acc2371d77..d99b6bc12b 100644
--- a/core-metadata/pom.xml
+++ b/core-metadata/pom.xml
@@ -24,6 +24,18 @@
<packaging>jar</packaging>
<name>Apache Kylin - Core Metadata</name>
<description>Apache Kylin - Core Metadata</description>
+ <build>
+ <plugins>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-compiler-plugin</artifactId>
+ <configuration>
+ <source>1.8</source>
+ <target>1.8</target>
+ </configuration>
+ </plugin>
+ </plugins>
+ </build>
<parent>
<groupId>org.apache.kylin</groupId>
@@ -87,6 +99,10 @@
<groupId>com.tdunning</groupId>
<artifactId>t-digest</artifactId>
</dependency>
+ <dependency>
+ <groupId>com.github.haifengl</groupId>
+ <artifactId>smile-core</artifactId>
+ </dependency>
<!-- Env & Test -->
<dependency>
diff --git a/core-metadata/src/main/java/org/apache/kylin/measure/MeasureTypeFactory.java b/core-metadata/src/main/java/org/apache/kylin/measure/MeasureTypeFactory.java
index 9699d2ea55..d18a8ac3ee 100644
--- a/core-metadata/src/main/java/org/apache/kylin/measure/MeasureTypeFactory.java
+++ b/core-metadata/src/main/java/org/apache/kylin/measure/MeasureTypeFactory.java
@@ -23,6 +23,7 @@
import org.apache.kylin.common.KylinConfig;
import org.apache.kylin.common.KylinConfigCannotInitException;
+import org.apache.kylin.measure.auc.AucMesureType;
import org.apache.kylin.measure.basic.BasicMeasureType;
import org.apache.kylin.measure.bitmap.BitmapMeasureType;
import org.apache.kylin.measure.dim.DimCountDistinctMeasureType;
@@ -112,6 +113,7 @@ public static synchronized void init() {
factoryInsts.add(new ExtendedColumnMeasureType.Factory());
factoryInsts.add(new PercentileMeasureType.Factory());
factoryInsts.add(new DimCountDistinctMeasureType.Factory());
+ factoryInsts.add(new AucMesureType.Factory());
logger.info("Checking custom measure types from kylin config");
diff --git a/core-metadata/src/main/java/org/apache/kylin/measure/auc/AucAggFunc.java b/core-metadata/src/main/java/org/apache/kylin/measure/auc/AucAggFunc.java
new file mode 100644
index 0000000000..018f371362
--- /dev/null
+++ b/core-metadata/src/main/java/org/apache/kylin/measure/auc/AucAggFunc.java
@@ -0,0 +1,51 @@
+/*
+ * 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.measure.auc;
+
+import org.apache.kylin.measure.ParamAsMeasureCount;
+
+public class AucAggFunc implements ParamAsMeasureCount {
+
+ public static AucCounter init() {
+ return null;
+ }
+
+ public static AucCounter add(AucCounter counter, Object t, Object p) {
+ if (counter == null) {
+ counter = new AucCounter();
+ }
+ counter.addTruth(t);
+ counter.addPred(p);
+ return counter;
+ }
+
+ public static AucCounter merge(AucCounter counter0, AucCounter counter1) {
+ counter0.merge(counter1);
+ return counter0;
+ }
+
+ public static double result(AucCounter counter) {
+ return counter == null ? -1D : counter.auc();
+ }
+
+ @Override
+ public int getParamAsMeasureCount() {
+ return 2;
+ }
+}
diff --git a/core-metadata/src/main/java/org/apache/kylin/measure/auc/AucAggregator.java b/core-metadata/src/main/java/org/apache/kylin/measure/auc/AucAggregator.java
new file mode 100644
index 0000000000..3b8edaa1d4
--- /dev/null
+++ b/core-metadata/src/main/java/org/apache/kylin/measure/auc/AucAggregator.java
@@ -0,0 +1,62 @@
+/*
+ * 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.measure.auc;
+
+import org.apache.kylin.measure.MeasureAggregator;
+
+public class AucAggregator extends MeasureAggregator<AucCounter> {
+ AucCounter auc = null;
+
+ public AucAggregator() {
+ }
+
+ @Override
+ public void reset() {
+ auc = null;
+ }
+
+ @Override
+ public void aggregate(AucCounter value) {
+ if (auc == null)
+ auc = new AucCounter(value);
+ else
+ auc.merge(value);
+ }
+
+ @Override
+ public AucCounter aggregate(AucCounter value1, AucCounter value2) {
+ if (value1 == null) {
+ return value2;
+ } else if (value2 == null) {
+ return value1;
+ }
+ value1.merge(value2);
+ return value1;
+ }
+
+ @Override
+ public AucCounter getState() {
+ return auc;
+ }
+
+ @Override
+ public int getMemBytesEstimate() {
+ return auc.getTruth().size() * 4 + auc.getPred().size() * 4;
+ }
+}
diff --git a/core-metadata/src/main/java/org/apache/kylin/measure/auc/AucCounter.java b/core-metadata/src/main/java/org/apache/kylin/measure/auc/AucCounter.java
new file mode 100644
index 0000000000..fd94bcaa5b
--- /dev/null
+++ b/core-metadata/src/main/java/org/apache/kylin/measure/auc/AucCounter.java
@@ -0,0 +1,96 @@
+/*
+ * 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.measure.auc;
+
+
+import com.google.common.collect.Lists;
+import org.apache.commons.collections.CollectionUtils;
+import smile.validation.AUC;
+
+import java.io.Serializable;
+import java.util.List;
+
+public class AucCounter implements Serializable {
+ private List<Integer> truth = Lists.newLinkedList();
+ private List<Double> pred = Lists.newLinkedList();
+
+ public AucCounter() {
+ }
+
+ public AucCounter(AucCounter another) {
+ merge(another);
+ }
+
+ public AucCounter(List<Integer> truth, List<Double> pred) {
+ this.truth = truth == null ? Lists.newLinkedList() : truth;
+ this.pred = pred == null ? Lists.newLinkedList() : pred;
+ }
+
+
+ public void merge(AucCounter value) {
+
+ if (value == null) {
+ return;
+ }
+
+ if (CollectionUtils.isEmpty(value.getTruth()) || CollectionUtils.isEmpty(value.getPred())) {
+ return;
+ }
+
+ this.getTruth().addAll(value.getTruth());
+ this.getPred().addAll(value.getPred());
+ }
+
+ public List<Integer> getTruth() {
+ return truth;
+ }
+
+ public List<Double> getPred() {
+ return pred;
+ }
+
+ public double auc() {
+ if (CollectionUtils.isEmpty(truth) || CollectionUtils.isEmpty(pred)) {
+ return -1;
+ }
+
+ int[] t = truth.stream().mapToInt(Integer::valueOf).toArray();
+ double[] p = pred.stream().mapToDouble(Double::valueOf).toArray();
+ double result = AUC.measure(t, p);
+ if (Double.isNaN(result)) {
+ return -1;
+ }
+ return result;
+ }
+
+ public void addTruth(Object t) {
+
+ if (t == null) {
+ throw new RuntimeException("Truth of dimension is null ");
+ }
+ truth.add((Integer) t);
+ }
+
+ public void addPred(Object p) {
+ if (p == null) {
+ throw new RuntimeException("Pred of dimension is null ");
+ }
+ pred.add((Double) p);
+ }
+}
diff --git a/core-metadata/src/main/java/org/apache/kylin/measure/auc/AucMesureType.java b/core-metadata/src/main/java/org/apache/kylin/measure/auc/AucMesureType.java
new file mode 100644
index 0000000000..0ec5ea2183
--- /dev/null
+++ b/core-metadata/src/main/java/org/apache/kylin/measure/auc/AucMesureType.java
@@ -0,0 +1,88 @@
+/*
+ * 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.measure.auc;
+
+
+import org.apache.kylin.measure.MeasureAggregator;
+import org.apache.kylin.measure.MeasureIngester;
+import org.apache.kylin.measure.MeasureType;
+import org.apache.kylin.measure.MeasureTypeFactory;
+
+import com.google.common.collect.ImmutableMap;
+import org.apache.kylin.metadata.datatype.DataType;
+import org.apache.kylin.metadata.datatype.DataTypeSerializer;
+
+import java.util.Map;
+
+public class AucMesureType extends MeasureType<AucCounter> {
+
+ private final DataType dataType;
+ public static final String FUNC_AUC = "AUC";
+ public static final String DATATYPE_AUC = "auc";
+
+ public AucMesureType(String funcName, DataType dataType) {
+ this.dataType = dataType;
+ }
+
+ public static class Factory extends MeasureTypeFactory<AucCounter> {
+
+ @Override
+ public MeasureType<AucCounter> createMeasureType(String funcName, DataType dataType) {
+ return new AucMesureType(funcName, dataType);
+ }
+
+ @Override
+ public String getAggrFunctionName() {
+ return FUNC_AUC;
+ }
+
+ @Override
+ public String getAggrDataTypeName() {
+ return DATATYPE_AUC;
+ }
+
+ @Override
+ public Class<? extends DataTypeSerializer<AucCounter>> getAggrDataTypeSerializer() {
+ return AucSerializer.class;
+ }
+ }
+
+ @Override
+ public MeasureIngester<AucCounter> newIngester() {
+ throw new UnsupportedOperationException("No ingester for this measure type.");
+ }
+
+ @Override
+ public MeasureAggregator<AucCounter> newAggregator() {
+ return new AucAggregator();
+ }
+
+ static final Map<String, Class<?>> UDAF_MAP = ImmutableMap.<String, Class<?>>of(
+ AucMesureType.FUNC_AUC, AucAggFunc.class);
+
+ @Override
+ public Map<String, Class<?>> getRewriteCalciteAggrFunctions() {
+ return UDAF_MAP;
+ }
+
+ @Override
+ public boolean needRewrite() {
+ return true;
+ }
+}
diff --git a/core-metadata/src/main/java/org/apache/kylin/measure/auc/AucSerializer.java b/core-metadata/src/main/java/org/apache/kylin/measure/auc/AucSerializer.java
new file mode 100644
index 0000000000..b13f14bbd8
--- /dev/null
+++ b/core-metadata/src/main/java/org/apache/kylin/measure/auc/AucSerializer.java
@@ -0,0 +1,83 @@
+/*
+ * 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.measure.auc;
+
+import org.apache.kylin.metadata.datatype.DataType;
+import org.apache.kylin.metadata.datatype.DataTypeSerializer;
+import org.apache.kylin.util.KryoUtils;
+
+import java.nio.ByteBuffer;
+import java.util.LinkedList;
+import java.util.List;
+
+public class AucSerializer extends DataTypeSerializer<AucCounter> {
+
+
+ // called by reflection
+ public AucSerializer(DataType type) {
+ }
+
+
+ @Override
+ public void serialize(AucCounter value, ByteBuffer out) {
+ byte[] tserialize = KryoUtils.serialize(value.getTruth());
+ byte[] pserialize = KryoUtils.serialize(value.getPred());
+ out.putInt(4 + 4 + tserialize.length);
+ out.putInt(tserialize.length);
+ out.put(tserialize);
+ out.putInt(4 + 4 + pserialize.length);
+ out.putInt(pserialize.length);
+ out.put(pserialize);
+
+ }
+
+ @Override
+ public AucCounter deserialize(ByteBuffer in) {
+ int totalTruthLength = in.getInt();
+ int tarrayLength = in.getInt();
+ byte[] tdata = new byte[tarrayLength];
+ in.get(tdata);
+ List<Integer> truth = KryoUtils.deserialize(tdata, LinkedList.class);
+
+ int totalpredLength = in.getInt();
+ int parrayLength = in.getInt();
+ byte[] pdata = new byte[parrayLength];
+ in.get(pdata);
+ List<Double> pred = KryoUtils.deserialize(pdata, LinkedList.class);
+ return new AucCounter(truth, pred);
+ }
+
+ @Override
+ public int peekLength(ByteBuffer in) {
+ int mark = in.position();
+ int ret = in.getInt();
+ in.position(mark);
+ return ret;
+ }
+
+ @Override
+ public int maxLength() {
+ return 8 * 1024 * 1024;
+ }
+
+ @Override
+ public int getStorageBytesEstimate() {
+ return 1024;
+ }
+}
diff --git a/core-metadata/src/main/java/org/apache/kylin/util/KryoUtils.java b/core-metadata/src/main/java/org/apache/kylin/util/KryoUtils.java
index 4059e9475c..76ac538ca9 100644
--- a/core-metadata/src/main/java/org/apache/kylin/util/KryoUtils.java
+++ b/core-metadata/src/main/java/org/apache/kylin/util/KryoUtils.java
@@ -19,9 +19,11 @@
package org.apache.kylin.util;
import java.util.BitSet;
+import java.util.LinkedList;
-import org.objenesis.strategy.StdInstantiatorStrategy;
+import org.objenesis.strategy.StdInstantiatorStrategy;
+import com.esotericsoftware.kryo.serializers.CollectionSerializer;
import com.esotericsoftware.kryo.Kryo;
import com.esotericsoftware.kryo.io.Input;
import com.esotericsoftware.kryo.io.Output;
@@ -53,6 +55,9 @@ public static Kryo getKryo() {
Kryo kryo = new Kryo();
kryo.setInstantiatorStrategy(new Kryo.DefaultInstantiatorStrategy(new StdInstantiatorStrategy()));
kryo.register(BitSet.class, new BitSetSerializer());
+ CollectionSerializer listSerializer = new CollectionSerializer();
+ listSerializer.setElementsCanBeNull(false);
+ kryo.register(LinkedList.class, listSerializer);
_Kryo.set(kryo);
}
diff --git a/core-metadata/src/test/java/org/apache/kylin/measure/auc/AucAggregatorTest.java b/core-metadata/src/test/java/org/apache/kylin/measure/auc/AucAggregatorTest.java
new file mode 100644
index 0000000000..036c67f7aa
--- /dev/null
+++ b/core-metadata/src/test/java/org/apache/kylin/measure/auc/AucAggregatorTest.java
@@ -0,0 +1,48 @@
+package org.apache.kylin.measure.auc;
+
+import com.google.common.collect.Lists;
+import org.apache.commons.math3.random.RandomDataGenerator;
+import org.junit.Test;
+import smile.validation.AUC;
+
+import java.util.List;
+
+import static org.junit.Assert.assertEquals;
+
+public class AucAggregatorTest {
+ @Test
+ public void testAggregate() {
+ int datasize = 10000;
+ AucAggregator aggregator = new AucAggregator();
+ RandomDataGenerator randomData = new RandomDataGenerator();
+ List<Integer> truths = Lists.newArrayListWithCapacity(datasize);
+ List<Double> preds = Lists.newArrayListWithCapacity(datasize);
+ for (int i = 0; i < datasize; i++) {
+ int t = randomData.nextInt(0, 1);
+ double p = Math.random();
+ truths.add(t);
+ preds.add(p);
+
+ AucCounter c = new AucCounter();
+ c.addTruth(t);
+ c.addPred(p);
+ aggregator.aggregate(c);
+ }
+
+ double actualResult = aggregator.getState().auc();
+ double expectResult = auc(truths, preds);
+ assertEquals(expectResult, actualResult, 0.001);
+ }
+
+
+ public double auc(List<Integer> truth, List<Double> pred) {
+
+ int[] t = truth.stream().mapToInt(Integer::valueOf).toArray();
+ double[] p = pred.stream().mapToDouble(Double::valueOf).toArray();
+ double result = AUC.measure(t, p);
+ if (Double.isNaN(result)) {
+ return -1;
+ }
+ return result;
+ }
+}
diff --git a/core-metadata/src/test/java/org/apache/kylin/measure/auc/AucSerializerTest.java b/core-metadata/src/test/java/org/apache/kylin/measure/auc/AucSerializerTest.java
new file mode 100644
index 0000000000..0c21fa896d
--- /dev/null
+++ b/core-metadata/src/test/java/org/apache/kylin/measure/auc/AucSerializerTest.java
@@ -0,0 +1,75 @@
+package org.apache.kylin.measure.auc;
+
+import org.apache.kylin.common.util.LocalFileMetadataTestCase;
+import org.apache.kylin.metadata.datatype.DataType;
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+import java.nio.ByteBuffer;
+import java.util.Arrays;
+import java.util.LinkedList;
+import java.util.List;
+
+import static org.junit.Assert.assertEquals;
+
+public class AucSerializerTest extends LocalFileMetadataTestCase {
+
+ @BeforeClass
+ public static void setUp() throws Exception {
+ staticCreateTestMetadata();
+ }
+
+ @AfterClass
+ public static void after() throws Exception {
+ cleanAfterClass();
+ }
+
+
+ @Test
+ public void testBasic() {
+ AucSerializer serializer = new AucSerializer(DataType.getType("auc"));
+
+ List<Integer> truth = new LinkedList<>(Arrays.asList(0, 0, 1, 1));
+ List<Double> pred = new LinkedList<>(Arrays.asList(0.1, 0.4, 0.35, 0.8));
+
+ AucCounter counter = new AucCounter(truth, pred);
+ double markResult = counter.auc();
+ assertEquals(markResult, 0.75, 0.01);
+
+ ByteBuffer buffer = ByteBuffer.allocateDirect(serializer.getStorageBytesEstimate());
+ serializer.serialize(counter, buffer);
+
+ buffer.flip();
+ counter = serializer.deserialize(buffer);
+ List<Integer> truth1 = new LinkedList<>(Arrays.asList(1, 0, 1, 1));
+ List<Double> pred1 = new LinkedList<>(Arrays.asList(0.9, 0.4, 0.65, 0.8));
+ AucCounter counter1 = new AucCounter(truth1, pred1);
+ counter1.merge(counter);
+
+ assertEquals(0.86, counter1.auc(), 0.01);
+ }
+
+ @Test
+ public void testNull() {
+
+ List<Integer> truth = null;
+ List<Double> pred = null;
+
+ AucCounter counter = new AucCounter(truth, pred);
+ double markResult = counter.auc();
+ assertEquals(markResult, -1.0, 0.01);
+
+ }
+
+ @Test
+ public void testNan() {
+
+ List<Integer> truth = new LinkedList<>(Arrays.asList(1, 1, 1, 1));
+ List<Double> pred = new LinkedList<>(Arrays.asList(0.1, 0.4, 0.35, 0.8));
+
+ AucCounter counter = new AucCounter(truth, pred);
+ double markResult = counter.auc();
+ assertEquals(markResult, -1.0, 0.01);
+ }
+}
diff --git a/pom.xml b/pom.xml
index d9b9efe20b..e6608ad5b2 100644
--- a/pom.xml
+++ b/pom.xml
@@ -125,6 +125,7 @@
<spring.framework.security.version>4.2.3.RELEASE</spring.framework.security.version>
<spring.framework.security.extensions.version>1.0.2.RELEASE</spring.framework.security.extensions.version>
<opensaml.version>2.6.6</opensaml.version>
+ <smile.version>1.5.0</smile.version>
<aspectj.version>1.8.9</aspectj.version>
<!-- Sonar -->
@@ -805,6 +806,11 @@
<artifactId>opensaml</artifactId>
<version>${opensaml.version}</version>
</dependency>
+ <dependency>
+ <groupId>com.github.haifengl</groupId>
+ <artifactId>smile-core</artifactId>
+ <version>${smile.version}</version>
+ </dependency>
<!-- Spring Core -->
diff --git a/storage-hbase/pom.xml b/storage-hbase/pom.xml
index c1b4cea76c..2ce142ece8 100644
--- a/storage-hbase/pom.xml
+++ b/storage-hbase/pom.xml
@@ -141,6 +141,7 @@
<include>org.apache.kylin:kylin-core-cube</include>
<include>org.roaringbitmap:RoaringBitmap</include>
<include>com.tdunning:t-digest</include>
+ <include>com.github.haifengl:smile-core</include>
</includes>
</artifactSet>
<relocations>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services