KYLIN-3514 Support AUC UDAF

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>

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");
 

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;
+    }
+}

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;
+    }
+}

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);
+    }
+}

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;
+    }
+}