You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hivemall.apache.org by my...@apache.org on 2019/11/22 14:17:18 UTC
[incubator-hivemall] branch master updated: [HIVEMALL-281] Support
max_by, min_by, majority_vote UDAFs
This is an automated email from the ASF dual-hosted git repository.
myui pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-hivemall.git
The following commit(s) were added to refs/heads/master by this push:
new 86409b4 [HIVEMALL-281] Support max_by, min_by, majority_vote UDAFs
86409b4 is described below
commit 86409b4347582e3de1c02fc99a20e8d7bb7706cf
Author: Makoto Yui <my...@apache.org>
AuthorDate: Fri Nov 22 23:17:11 2019 +0900
[HIVEMALL-281] Support max_by, min_by, majority_vote UDAFs
## What changes were proposed in this pull request?
upport max_by, min_by, majority_vote UDAFs
## What type of PR is it?
Feature
## What is the Jira issue?
https://issues.apache.org/jira/browse/HIVEMALL-281
## How was this patch tested?
manual tests on EMR
## How to use this feature?
```sql
create table data1 as (
select 'jake' as name, 18 as age
union all
select 'tom' as name, 64 as age
union all
select 'lisa' as name, 32 as age
);
select
max_by(name, age) as max_name,
min_by(name, age) as min_name
from
data1;
> tom, jake
create table data2 as
select
explode(array('1', '2', '2', '2', '5', '4', '1', '2')) as k;
select
majority_vote(k) as k
from
data2;
> 2
```
## Checklist
(Please remove this section if not needed; check `x` for YES, blank for NO)
- [x] Did you apply source code formatter, i.e., `./bin/format_code.sh`, for your commit?
- [x] Did you run system tests on Hive (or Spark)?
Author: Makoto Yui <my...@apache.org>
Closes #214 from myui/HIVEMALL-281.
---
.../java/hivemall/tools/aggr/MajorityVoteUDAF.java | 232 +++++++++++++++++++++
.../main/java/hivemall/tools/aggr/MaxByUDAF.java | 214 +++++++++++++++++++
.../main/java/hivemall/tools/aggr/MinByUDAF.java | 215 +++++++++++++++++++
.../main/java/hivemall/utils/lang/LongCounter.java | 125 +++++++++++
docs/gitbook/misc/funcs.md | 2 +-
docs/gitbook/misc/generic_funcs.md | 51 +++++
resources/ddl/define-all-as-permanent.hive | 13 ++
resources/ddl/define-all.hive | 14 ++
resources/ddl/define-all.spark | 13 +-
.../java/hivemall/docs/FuncsListGeneratorMojo.java | 1 +
10 files changed, 878 insertions(+), 2 deletions(-)
diff --git a/core/src/main/java/hivemall/tools/aggr/MajorityVoteUDAF.java b/core/src/main/java/hivemall/tools/aggr/MajorityVoteUDAF.java
new file mode 100644
index 0000000..89f0409
--- /dev/null
+++ b/core/src/main/java/hivemall/tools/aggr/MajorityVoteUDAF.java
@@ -0,0 +1,232 @@
+/*
+ * 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 hivemall.tools.aggr;
+
+import static hivemall.utils.hadoop.HiveUtils.asLongOI;
+import static hivemall.utils.hadoop.HiveUtils.asPrimitiveObjectInspector;
+
+import hivemall.utils.hadoop.HiveUtils;
+import hivemall.utils.lang.LongCounter;
+import hivemall.utils.lang.Preconditions;
+
+import java.util.Map;
+
+import javax.annotation.Nonnull;
+import javax.annotation.Nullable;
+
+import org.apache.hadoop.hive.ql.exec.Description;
+import org.apache.hadoop.hive.ql.exec.UDFArgumentLengthException;
+import org.apache.hadoop.hive.ql.exec.UDFArgumentTypeException;
+import org.apache.hadoop.hive.ql.metadata.HiveException;
+import org.apache.hadoop.hive.ql.parse.SemanticException;
+import org.apache.hadoop.hive.ql.udf.generic.AbstractGenericUDAFResolver;
+import org.apache.hadoop.hive.ql.udf.generic.GenericUDAFEvaluator;
+import org.apache.hadoop.hive.ql.udf.generic.GenericUDAFEvaluator.AbstractAggregationBuffer;
+import org.apache.hadoop.hive.ql.udf.generic.GenericUDAFEvaluator.AggregationType;
+import org.apache.hadoop.hive.ql.util.JavaDataModel;
+import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector;
+import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspectorFactory;
+import org.apache.hadoop.hive.serde2.objectinspector.PrimitiveObjectInspector;
+import org.apache.hadoop.hive.serde2.objectinspector.StandardMapObjectInspector;
+import org.apache.hadoop.hive.serde2.objectinspector.primitive.LongObjectInspector;
+import org.apache.hadoop.hive.serde2.objectinspector.primitive.PrimitiveObjectInspectorFactory;
+import org.apache.hadoop.hive.serde2.typeinfo.TypeInfo;
+
+//@formatter:off
+@Description(name = "majority_vote",
+ value = "_FUNC_(Primitive x) - Returns the most frequent value of x",
+ extended = "-- see https://issues.apache.org/jira/browse/HIVE-17406 \n" +
+ "WITH data as (\n" +
+ " select\n" +
+ " explode(array('1', '2', '2', '2', '5', '4', '1', '2')) as k\n" +
+ ")\n" +
+ "select\n" +
+ " majority_vote(k) as k\n" +
+ "from \n" +
+ " data;\n" +
+ "> 2")
+//@formatter:on
+public final class MajorityVoteUDAF extends AbstractGenericUDAFResolver {
+
+ public MajorityVoteUDAF() {
+ super();
+ }
+
+ @Override
+ public GenericUDAFEvaluator getEvaluator(@Nonnull final TypeInfo[] argTypes)
+ throws SemanticException {
+ if (argTypes.length != 1) {
+ throw new UDFArgumentLengthException(
+ "Expected ecactly one argument: " + argTypes.length);
+ }
+ if (!HiveUtils.isPrimitiveTypeInfo(argTypes[0])) {
+ throw new UDFArgumentTypeException(0,
+ "PRIMITIVE type is expected but got " + argTypes[0]);
+ }
+ return new Evaluator();
+ }
+
+ public static final class Evaluator extends GenericUDAFEvaluator {
+
+ // original input
+ private transient PrimitiveObjectInspector keyInputOI;
+ private transient PrimitiveObjectInspector keyOutputOI;
+
+ // partial aggregation
+ private transient StandardMapObjectInspector partialOI;
+ private transient LongObjectInspector counterInputOI;
+
+ public Evaluator() {
+ super();
+ }
+
+ @Override
+ public ObjectInspector init(@Nonnull Mode mode, @Nonnull ObjectInspector[] argOIs)
+ throws HiveException {
+ assert (argOIs.length == 1);
+ super.init(mode, argOIs);
+
+ // initialize input
+ if (mode == Mode.PARTIAL1 || mode == Mode.COMPLETE) {// from original data
+ this.keyInputOI = asPrimitiveObjectInspector(argOIs[0]);
+ } else {// from partial aggregation
+ this.partialOI = (StandardMapObjectInspector) argOIs[0];
+ this.keyInputOI = asPrimitiveObjectInspector(partialOI.getMapKeyObjectInspector());
+ this.counterInputOI = asLongOI(partialOI.getMapValueObjectInspector());
+ }
+ this.keyOutputOI = PrimitiveObjectInspectorFactory.getPrimitiveJavaObjectInspector(
+ keyInputOI.getTypeInfo());
+
+ // initialize output
+ final ObjectInspector outputOI;
+ if (mode == Mode.PARTIAL1 || mode == Mode.PARTIAL2) {// terminatePartial
+ outputOI = ObjectInspectorFactory.getStandardMapObjectInspector(keyOutputOI,
+ PrimitiveObjectInspectorFactory.javaLongObjectInspector);
+ } else {// terminate
+ outputOI = keyOutputOI;
+ }
+ return outputOI;
+ }
+
+ @Override
+ public CounterBuffer getNewAggregationBuffer() throws HiveException {
+ CounterBuffer buf = new CounterBuffer();
+ buf.reset();
+ return buf;
+ }
+
+ @Override
+ public void reset(@SuppressWarnings("deprecation") AggregationBuffer agg)
+ throws HiveException {
+ CounterBuffer buf = (CounterBuffer) agg;
+ buf.reset();
+ }
+
+ @Override
+ public void iterate(@SuppressWarnings("deprecation") AggregationBuffer agg,
+ Object[] parameters) throws HiveException {
+ CounterBuffer buf = (CounterBuffer) agg;
+
+ final Object param0 = parameters[0];
+ if (param0 == null) {
+ return;
+ }
+
+ final Object key = keyInputOI.getPrimitiveJavaObject(param0);
+ if (key != null) {
+ buf.iterate(key);
+ }
+ }
+
+ @Override
+ public Map<Object, Long> terminatePartial(
+ @SuppressWarnings("deprecation") AggregationBuffer agg) throws HiveException {
+ CounterBuffer buf = (CounterBuffer) agg;
+
+ return buf.terminatePartial();
+ }
+
+ @Override
+ public void merge(@SuppressWarnings("deprecation") AggregationBuffer agg, Object partial)
+ throws HiveException {
+ if (partial == null) {
+ return;
+ }
+
+ final CounterBuffer buf = (CounterBuffer) agg;
+ Map<?, ?> partialResult = partialOI.getMap(partial);
+ for (Map.Entry<?, ?> e : partialResult.entrySet()) {
+ Object k =
+ keyInputOI.getPrimitiveJavaObject(Preconditions.checkNotNull(e.getKey()));
+ long v = counterInputOI.get(Preconditions.checkNotNull(e.getValue()));
+ buf.merge(k, v);
+ }
+ }
+
+ @Override
+ public Object terminate(@SuppressWarnings("deprecation") AggregationBuffer agg)
+ throws HiveException {
+ CounterBuffer buf = (CounterBuffer) agg;
+ return buf.terminate();
+ }
+
+ }
+
+ @AggregationType(estimable = true)
+ public static final class CounterBuffer extends AbstractAggregationBuffer {
+
+ @Nonnull
+ private LongCounter<Object> partial;
+
+ public CounterBuffer() {
+ super();
+ reset();
+ }
+
+ void reset() {
+ this.partial = new LongCounter<Object>();
+ }
+
+ void iterate(@Nonnull final Object k) {
+ partial.increment(k);
+ }
+
+ void merge(@Nonnull final Object k, final long v) {
+ partial.increment(k, v);
+ }
+
+ @Nonnull
+ Map<Object, Long> terminatePartial() {
+ return partial.getMap();
+ }
+
+ @Nullable
+ Object terminate() {
+ return partial.whichMax();
+ }
+
+ @Override
+ public int estimate() {
+ int size = partial.size();
+ return JavaDataModel.PRIMITIVES2 * size * 2; // rough estimate
+ }
+
+ }
+
+}
diff --git a/core/src/main/java/hivemall/tools/aggr/MaxByUDAF.java b/core/src/main/java/hivemall/tools/aggr/MaxByUDAF.java
new file mode 100644
index 0000000..ae89a70
--- /dev/null
+++ b/core/src/main/java/hivemall/tools/aggr/MaxByUDAF.java
@@ -0,0 +1,214 @@
+/*
+ * 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 hivemall.tools.aggr;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import javax.annotation.Nonnull;
+import javax.annotation.Nullable;
+
+import org.apache.hadoop.hive.ql.exec.Description;
+import org.apache.hadoop.hive.ql.exec.UDFArgumentLengthException;
+import org.apache.hadoop.hive.ql.exec.UDFArgumentTypeException;
+import org.apache.hadoop.hive.ql.metadata.HiveException;
+import org.apache.hadoop.hive.ql.parse.SemanticException;
+import org.apache.hadoop.hive.ql.udf.UDFType;
+import org.apache.hadoop.hive.ql.udf.generic.AbstractGenericUDAFResolver;
+import org.apache.hadoop.hive.ql.udf.generic.GenericUDAFEvaluator;
+import org.apache.hadoop.hive.ql.util.JavaDataModel;
+import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector;
+import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspectorFactory;
+import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspectorUtils;
+import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspectorUtils.ObjectInspectorCopyOption;
+import org.apache.hadoop.hive.serde2.objectinspector.StructField;
+import org.apache.hadoop.hive.serde2.objectinspector.StructObjectInspector;
+import org.apache.hadoop.hive.serde2.typeinfo.TypeInfo;
+import org.apache.hadoop.hive.serde2.typeinfo.TypeInfoUtils;
+
+//@formatter:off
+@Description(name = "max_by",
+ value = "_FUNC_(x, y) - Returns the value of x associated with the maximum value of y over all input values.",
+ extended = "-- see https://issues.apache.org/jira/browse/HIVE-17406 \n"
+ + "WITH data as (\n" +
+ " select 'jake' as name, 18 as age\n" +
+ " union all\n" +
+ " select 'tom' as name, 64 as age\n" +
+ " union all\n" +
+ " select 'lisa' as name, 32 as age\n" +
+ ")\n" +
+ "select\n" +
+ " max_by(name, age) as name\n" +
+ "from\n" +
+ " data;\n" +
+ "> tom")
+//@formatter:on
+public final class MaxByUDAF extends AbstractGenericUDAFResolver {
+
+ @Override
+ public GenericUDAFEvaluator getEvaluator(@Nonnull TypeInfo[] argTypes)
+ throws SemanticException {
+ if (argTypes.length != 2) {
+ throw new UDFArgumentLengthException(
+ "Exactly two arguments are expected: " + argTypes.length);
+ }
+ ObjectInspector yOI = TypeInfoUtils.getStandardJavaObjectInspectorFromTypeInfo(argTypes[1]);
+ if (!ObjectInspectorUtils.compareSupported(yOI)) {
+ throw new UDFArgumentTypeException(1,
+ "Cannot support comparison of map<> type or complex type containing map<>.");
+ }
+ return new Evaluator();
+ }
+
+ @UDFType(distinctLike = true)
+ public static class Evaluator extends GenericUDAFEvaluator {
+
+ private transient ObjectInspector xInputOI, yInputOI;
+ private transient ObjectInspector xOutputOI, yOutputOI;
+
+ @Nullable
+ private transient StructField xField, yField;
+ @Nullable
+ private transient StructObjectInspector partialInputOI;
+
+ @Override
+ public ObjectInspector init(Mode mode, ObjectInspector[] argOIs) throws HiveException {
+ super.init(mode, argOIs);
+
+ // initialize input
+ if (mode == Mode.PARTIAL1 || mode == Mode.COMPLETE) {// from original data
+ this.xInputOI = argOIs[0];
+ this.yInputOI = argOIs[1];
+ if (!ObjectInspectorUtils.compareSupported(yInputOI)) {
+ throw new UDFArgumentTypeException(1,
+ "Cannot support comparison of map<> type or complex type containing map<>.");
+ }
+ } else {// from partial aggregation
+ this.partialInputOI = (StructObjectInspector) argOIs[0];
+ this.xField = partialInputOI.getStructFieldRef("x");
+ this.xInputOI = xField.getFieldObjectInspector();
+ this.yField = partialInputOI.getStructFieldRef("y");
+ this.yInputOI = yField.getFieldObjectInspector();
+ }
+ this.xOutputOI = ObjectInspectorUtils.getStandardObjectInspector(xInputOI,
+ ObjectInspectorCopyOption.JAVA);
+ this.yOutputOI = ObjectInspectorUtils.getStandardObjectInspector(yInputOI,
+ ObjectInspectorCopyOption.JAVA);
+
+ // initialize output
+ final ObjectInspector outputOI;
+ if (mode == Mode.PARTIAL1 || mode == Mode.PARTIAL2) {// terminatePartial
+ List<String> fieldNames = new ArrayList<>(2);
+ List<ObjectInspector> fieldOIs = new ArrayList<>(2);
+ fieldNames.add("x");
+ fieldOIs.add(xOutputOI);
+ fieldNames.add("y");
+ fieldOIs.add(yOutputOI);
+ return ObjectInspectorFactory.getStandardStructObjectInspector(fieldNames,
+ fieldOIs);
+ } else {// terminate
+ // Copy to Java object because that saves object creation time.
+ outputOI = ObjectInspectorUtils.getStandardObjectInspector(xInputOI,
+ ObjectInspectorCopyOption.JAVA);
+ }
+ return outputOI;
+ }
+
+ /** class for storing the current max value */
+ @AggregationType(estimable = true)
+ static class MaxAgg extends AbstractAggregationBuffer {
+ Object x, y;
+
+ @Override
+ public int estimate() {
+ return JavaDataModel.PRIMITIVES2 * 2; // rough estimate
+ }
+
+ void merge(final Object newX, final Object newY,
+ @Nonnull final ObjectInspector xInputOI,
+ @Nonnull final ObjectInspector yInputOI,
+ @Nonnull final ObjectInspector yOutputOI) {
+ final int cmp = ObjectInspectorUtils.compare(y, yOutputOI, newY, yInputOI);
+ if (x == null || cmp < 0) { // found greater y
+ this.x = ObjectInspectorUtils.copyToStandardObject(newX, xInputOI,
+ ObjectInspectorCopyOption.JAVA);
+ this.y = ObjectInspectorUtils.copyToStandardObject(newY, yInputOI,
+ ObjectInspectorCopyOption.JAVA);
+ }
+ }
+ }
+
+ @Override
+ public MaxAgg getNewAggregationBuffer() throws HiveException {
+ return new MaxAgg();
+ }
+
+ @Override
+ public void reset(@SuppressWarnings("deprecation") AggregationBuffer agg)
+ throws HiveException {
+ MaxAgg myagg = (MaxAgg) agg;
+ myagg.x = null;
+ myagg.y = null;
+ }
+
+ @Override
+ public void iterate(@SuppressWarnings("deprecation") AggregationBuffer agg,
+ Object[] parameters) throws HiveException {
+ assert (parameters.length == 2);
+ MaxAgg myagg = (MaxAgg) agg;
+ Object x = parameters[0];
+ Object y = parameters[1];
+
+ myagg.merge(x, y, xInputOI, yInputOI, yOutputOI);
+ }
+
+ @Override
+ public Object terminatePartial(@SuppressWarnings("deprecation") AggregationBuffer agg)
+ throws HiveException {
+ MaxAgg myagg = (MaxAgg) agg;
+ Object[] partial = new Object[2];
+ partial[0] = myagg.x;
+ partial[1] = myagg.y;
+ return partial;
+ }
+
+ @Override
+ public void merge(@SuppressWarnings("deprecation") AggregationBuffer agg, Object partial)
+ throws HiveException {
+ if (partial == null) {
+ return;
+ }
+
+ MaxAgg myagg = (MaxAgg) agg;
+ Object x = partialInputOI.getStructFieldData(partial, xField);
+ Object y = partialInputOI.getStructFieldData(partial, yField);
+
+ myagg.merge(x, y, xInputOI, yInputOI, yOutputOI);
+ }
+
+ @Override
+ public Object terminate(@SuppressWarnings("deprecation") AggregationBuffer agg)
+ throws HiveException {
+ MaxAgg myagg = (MaxAgg) agg;
+ return myagg.x;
+ }
+
+ }
+
+}
diff --git a/core/src/main/java/hivemall/tools/aggr/MinByUDAF.java b/core/src/main/java/hivemall/tools/aggr/MinByUDAF.java
new file mode 100644
index 0000000..1c001ea
--- /dev/null
+++ b/core/src/main/java/hivemall/tools/aggr/MinByUDAF.java
@@ -0,0 +1,215 @@
+/*
+ * 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 hivemall.tools.aggr;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import javax.annotation.Nonnull;
+import javax.annotation.Nullable;
+
+import org.apache.hadoop.hive.ql.exec.Description;
+import org.apache.hadoop.hive.ql.exec.UDFArgumentLengthException;
+import org.apache.hadoop.hive.ql.exec.UDFArgumentTypeException;
+import org.apache.hadoop.hive.ql.metadata.HiveException;
+import org.apache.hadoop.hive.ql.parse.SemanticException;
+import org.apache.hadoop.hive.ql.udf.UDFType;
+import org.apache.hadoop.hive.ql.udf.generic.AbstractGenericUDAFResolver;
+import org.apache.hadoop.hive.ql.udf.generic.GenericUDAFEvaluator;
+import org.apache.hadoop.hive.ql.util.JavaDataModel;
+import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector;
+import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspectorFactory;
+import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspectorUtils;
+import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspectorUtils.ObjectInspectorCopyOption;
+import org.apache.hadoop.hive.serde2.objectinspector.StructField;
+import org.apache.hadoop.hive.serde2.objectinspector.StructObjectInspector;
+import org.apache.hadoop.hive.serde2.typeinfo.TypeInfo;
+import org.apache.hadoop.hive.serde2.typeinfo.TypeInfoUtils;
+
+//@formatter:off
+@Description(name = "min_by",
+ value = "_FUNC_(x, y) - Returns the value of x associated with the minimum value of y over all input values.",
+ extended = "-- see https://issues.apache.org/jira/browse/HIVE-17406 \n" +
+ "WITH data as (\n" +
+ " select 'jake' as name, 18 as age\n" +
+ " union all\n" +
+ " select 'tom' as name, 64 as age\n" +
+ " union all\n" +
+ " select 'lisa' as name, 32 as age\n" +
+ ")\n" +
+ "select\n" +
+ " min_by(name, age) as name\n" +
+ "from\n" +
+ " data;\n" +
+ "\n" +
+ "> jake")
+//@formatter:on
+public final class MinByUDAF extends AbstractGenericUDAFResolver {
+
+ @Override
+ public GenericUDAFEvaluator getEvaluator(@Nonnull TypeInfo[] argTypes)
+ throws SemanticException {
+ if (argTypes.length != 2) {
+ throw new UDFArgumentLengthException(
+ "Exactly two arguments are expected: " + argTypes.length);
+ }
+ ObjectInspector yOI = TypeInfoUtils.getStandardJavaObjectInspectorFromTypeInfo(argTypes[1]);
+ if (!ObjectInspectorUtils.compareSupported(yOI)) {
+ throw new UDFArgumentTypeException(1,
+ "Cannot support comparison of map<> type or complex type containing map<>.");
+ }
+ return new Evaluator();
+ }
+
+ @UDFType(distinctLike = true)
+ public static class Evaluator extends GenericUDAFEvaluator {
+
+ private transient ObjectInspector xInputOI, yInputOI;
+ private transient ObjectInspector xOutputOI, yOutputOI;
+
+ @Nullable
+ private transient StructField xField, yField;
+ @Nullable
+ private transient StructObjectInspector partialInputOI;
+
+ @Override
+ public ObjectInspector init(Mode mode, ObjectInspector[] argOIs) throws HiveException {
+ super.init(mode, argOIs);
+
+ // initialize input
+ if (mode == Mode.PARTIAL1 || mode == Mode.COMPLETE) {// from original data
+ this.xInputOI = argOIs[0];
+ this.yInputOI = argOIs[1];
+ if (!ObjectInspectorUtils.compareSupported(yInputOI)) {
+ throw new UDFArgumentTypeException(1,
+ "Cannot support comparison of map<> type or complex type containing map<>.");
+ }
+ } else {// from partial aggregation
+ this.partialInputOI = (StructObjectInspector) argOIs[0];
+ this.xField = partialInputOI.getStructFieldRef("x");
+ this.xInputOI = xField.getFieldObjectInspector();
+ this.yField = partialInputOI.getStructFieldRef("y");
+ this.yInputOI = yField.getFieldObjectInspector();
+ }
+ this.xOutputOI = ObjectInspectorUtils.getStandardObjectInspector(xInputOI,
+ ObjectInspectorCopyOption.JAVA);
+ this.yOutputOI = ObjectInspectorUtils.getStandardObjectInspector(yInputOI,
+ ObjectInspectorCopyOption.JAVA);
+
+ // initialize output
+ final ObjectInspector outputOI;
+ if (mode == Mode.PARTIAL1 || mode == Mode.PARTIAL2) {// terminatePartial
+ List<String> fieldNames = new ArrayList<>(2);
+ List<ObjectInspector> fieldOIs = new ArrayList<>(2);
+ fieldNames.add("x");
+ fieldOIs.add(xOutputOI);
+ fieldNames.add("y");
+ fieldOIs.add(yOutputOI);
+ return ObjectInspectorFactory.getStandardStructObjectInspector(fieldNames,
+ fieldOIs);
+ } else {// terminate
+ // Copy to Java object because that saves object creation time.
+ outputOI = ObjectInspectorUtils.getStandardObjectInspector(xInputOI,
+ ObjectInspectorCopyOption.JAVA);
+ }
+ return outputOI;
+ }
+
+ /** class for storing the current max value */
+ @AggregationType(estimable = false)
+ static class MinAgg extends AbstractAggregationBuffer {
+ Object x, y;
+
+ @Override
+ public int estimate() {
+ return JavaDataModel.PRIMITIVES2 * 2; // rough estimate
+ }
+
+ void merge(final Object newX, final Object newY,
+ @Nonnull final ObjectInspector xInputOI,
+ @Nonnull final ObjectInspector yInputOI,
+ @Nonnull final ObjectInspector yOutputOI) {
+ final int cmp = ObjectInspectorUtils.compare(y, yOutputOI, newY, yInputOI);
+ if (x == null || cmp > 0) {// found smaller y
+ this.x = ObjectInspectorUtils.copyToStandardObject(newX, xInputOI,
+ ObjectInspectorCopyOption.JAVA);
+ this.y = ObjectInspectorUtils.copyToStandardObject(newY, yInputOI,
+ ObjectInspectorCopyOption.JAVA);
+ }
+ }
+ }
+
+ @Override
+ public MinAgg getNewAggregationBuffer() throws HiveException {
+ return new MinAgg();
+ }
+
+ @Override
+ public void reset(@SuppressWarnings("deprecation") AggregationBuffer agg)
+ throws HiveException {
+ MinAgg myagg = (MinAgg) agg;
+ myagg.x = null;
+ myagg.y = null;
+ }
+
+ @Override
+ public void iterate(@SuppressWarnings("deprecation") AggregationBuffer agg,
+ Object[] parameters) throws HiveException {
+ assert (parameters.length == 2);
+ MinAgg myagg = (MinAgg) agg;
+ Object x = parameters[0];
+ Object y = parameters[1];
+
+ myagg.merge(x, y, xInputOI, yInputOI, yOutputOI);
+ }
+
+ @Override
+ public Object terminatePartial(@SuppressWarnings("deprecation") AggregationBuffer agg)
+ throws HiveException {
+ MinAgg myagg = (MinAgg) agg;
+ Object[] partial = new Object[2];
+ partial[0] = myagg.x;
+ partial[1] = myagg.y;
+ return partial;
+ }
+
+ @Override
+ public void merge(@SuppressWarnings("deprecation") AggregationBuffer agg, Object partial)
+ throws HiveException {
+ if (partial == null) {
+ return;
+ }
+
+ MinAgg myagg = (MinAgg) agg;
+ Object x = partialInputOI.getStructFieldData(partial, xField);
+ Object y = partialInputOI.getStructFieldData(partial, yField);
+
+ myagg.merge(x, y, xInputOI, yInputOI, yOutputOI);
+ }
+
+ @Override
+ public Object terminate(@SuppressWarnings("deprecation") AggregationBuffer agg)
+ throws HiveException {
+ MinAgg myagg = (MinAgg) agg;
+ return myagg.x;
+ }
+
+ }
+
+}
diff --git a/core/src/main/java/hivemall/utils/lang/LongCounter.java b/core/src/main/java/hivemall/utils/lang/LongCounter.java
new file mode 100644
index 0000000..ab45c02
--- /dev/null
+++ b/core/src/main/java/hivemall/utils/lang/LongCounter.java
@@ -0,0 +1,125 @@
+/*
+ * 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 hivemall.utils.lang;
+
+import java.io.Serializable;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Set;
+
+import javax.annotation.Nonnull;
+import javax.annotation.Nullable;
+
+public final class LongCounter<E> implements Serializable {
+ private static final long serialVersionUID = 7949630590734361716L;
+
+ private final Map<E, Long> counts;
+
+ public LongCounter() {
+ this.counts = new HashMap<E, Long>();
+ }
+
+ public LongCounter(@Nonnull Map<E, Long> counts) {
+ this.counts = counts;
+ }
+
+ public Map<E, Long> getMap() {
+ return counts;
+ }
+
+ public long increment(E key) {
+ return increment(key, 1L);
+ }
+
+ public long increment(E key, long amount) {
+ Long count = counts.get(key);
+ if (count == null) {
+ counts.put(key, Long.valueOf(amount));
+ return 0;
+ } else {
+ long old = count.longValue();
+ counts.put(key, Long.valueOf(old + amount));
+ return old;
+ }
+ }
+
+ public long getCount(E key) {
+ Long count = counts.get(key);
+ if (count == null) {
+ return 0;
+ } else {
+ return count.longValue();
+ }
+ }
+
+ public void addAll(Map<E, Long> counter) {
+ if (counter == null) {
+ return;
+ }
+ for (Map.Entry<E, Long> e : counter.entrySet()) {
+ increment(e.getKey(), e.getValue().longValue());
+ }
+ }
+
+ public void addAll(LongCounter<E> counter) {
+ if (counter == null) {
+ return;
+ }
+ for (Map.Entry<E, Long> e : counter.entrySet()) {
+ increment(e.getKey(), e.getValue().longValue());
+ }
+ }
+
+ public Set<Map.Entry<E, Long>> entrySet() {
+ return counts.entrySet();
+ }
+
+ @Nullable
+ public E whichMax() {
+ E maxKey = null;
+ long maxValue = Long.MIN_VALUE;
+ for (Map.Entry<E, Long> e : counts.entrySet()) {
+ long v = e.getValue().longValue();
+ if (v >= maxValue) {
+ maxValue = v;
+ maxKey = e.getKey();
+ }
+ }
+ return maxKey;
+ }
+
+ @Nullable
+ public E whichMin() {
+ E minKey = null;
+ long minValue = Long.MAX_VALUE;
+ for (Map.Entry<E, Long> e : counts.entrySet()) {
+ long v = e.getValue().longValue();
+ if (v <= minValue) {
+ minValue = v;
+ minKey = e.getKey();
+ }
+ }
+ return minKey;
+ }
+
+ public int size() {
+ return counts.size();
+ }
+
+}
diff --git a/docs/gitbook/misc/funcs.md b/docs/gitbook/misc/funcs.md
index 13ecea5..0f19ff1 100644
--- a/docs/gitbook/misc/funcs.md
+++ b/docs/gitbook/misc/funcs.md
@@ -638,7 +638,7 @@ Reference: <a href="https://papers.nips.cc/paper/3848-adaptive-regularization-of
- `train_multiclass_xgboost_classifier(string[] features, double target [, string options])` - Returns a relation consisting of <string model_id, array<byte> pred_model>
-- `train_xgboost_classifier(string[] features, double target [, string options])` - Returns a relation consisting of <string model_id, array<byte> pred_model>
+- `train_xgboost_classifier(array<string> features, double target [, string options])` - Returns a relation consisting of <string model_id, array<byte> pred_model>
- `train_xgboost_regr(string[] features, double target [, string options])` - Returns a relation consisting of <string model_id, array<byte> pred_model>
diff --git a/docs/gitbook/misc/generic_funcs.md b/docs/gitbook/misc/generic_funcs.md
index 9dfbc2b..73983f5 100644
--- a/docs/gitbook/misc/generic_funcs.md
+++ b/docs/gitbook/misc/generic_funcs.md
@@ -21,6 +21,57 @@ This page describes a list of useful Hivemall generic functions. See also a [lis
<!-- toc -->
+# Aggregation
+
+- `majority_vote(Primitive x)` - Returns the most frequent value of x
+ ```sql
+ -- see https://issues.apache.org/jira/browse/HIVE-17406
+ WITH data as (
+ select
+ explode(array('1', '2', '2', '2', '5', '4', '1', '2')) as k
+ )
+ select
+ majority_vote(k) as k
+ from
+ data;
+ > 2
+ ```
+
+- `max_by(x, y)` - Returns the value of x associated with the maximum value of y over all input values.
+ ```sql
+ -- see https://issues.apache.org/jira/browse/HIVE-17406
+ WITH data as (
+ select 'jake' as name, 18 as age
+ union all
+ select 'tom' as name, 64 as age
+ union all
+ select 'lisa' as name, 32 as age
+ )
+ select
+ max_by(name, age) as name
+ from
+ data;
+ > tom
+ ```
+
+- `min_by(x, y)` - Returns the value of x associated with the minimum value of y over all input values.
+ ```sql
+ -- see https://issues.apache.org/jira/browse/HIVE-17406
+ WITH data as (
+ select 'jake' as name, 18 as age
+ union all
+ select 'tom' as name, 64 as age
+ union all
+ select 'lisa' as name, 32 as age
+ )
+ select
+ min_by(name, age) as name
+ from
+ data;
+
+ > jake
+ ```
+
# Array
- `arange([int start=0, ] int stop, [int step=1])` - Return evenly spaced values within a given interval
diff --git a/resources/ddl/define-all-as-permanent.hive b/resources/ddl/define-all-as-permanent.hive
index 6cbb7d2..c8bf307 100644
--- a/resources/ddl/define-all-as-permanent.hive
+++ b/resources/ddl/define-all-as-permanent.hive
@@ -883,6 +883,19 @@ CREATE FUNCTION bloom_or as 'hivemall.sketch.bloom.BloomOrUDF' USING JAR '${hive
DROP FUNCTION IF EXISTS bloom_contains_any;
CREATE FUNCTION bloom_contains_any as 'hivemall.sketch.bloom.BloomContainsAnyUDF' USING JAR '${hivemall_jar}';
+-----------------
+-- Aggregation --
+-----------------
+
+DROP FUNCTION IF EXISTS max_by;
+CREATE FUNCTION max_by as 'hivemall.tools.aggr.MaxByUDAF' USING JAR '${hivemall_jar}';
+
+DROP FUNCTION IF EXISTS min_by;
+CREATE FUNCTION min_by as 'hivemall.tools.aggr.MinByUDAF' USING JAR '${hivemall_jar}';
+
+DROP FUNCTION IF EXISTS majority_vote;
+CREATE FUNCTION majority_vote as 'hivemall.tools.aggr.MajorityVoteUDAF' USING JAR '${hivemall_jar}';
+
------------------------------
-- XGBoost related features --
------------------------------
diff --git a/resources/ddl/define-all.hive b/resources/ddl/define-all.hive
index 4efae3a..f8461ef 100644
--- a/resources/ddl/define-all.hive
+++ b/resources/ddl/define-all.hive
@@ -875,6 +875,20 @@ create temporary function bloom_or as 'hivemall.sketch.bloom.BloomOrUDF';
drop temporary function if exists bloom_contains_any;
create temporary function bloom_contains_any as 'hivemall.sketch.bloom.BloomContainsAnyUDF';
+-----------------
+-- Aggregation --
+-----------------
+
+drop temporary function if exists max_by;
+create temporary function max_by as 'hivemall.tools.aggr.MaxByUDAF';
+
+drop temporary function if exists min_by;
+create temporary function min_by as 'hivemall.tools.aggr.MinByUDAF';
+
+drop temporary function if exists majority_vote;
+create temporary function majority_vote as 'hivemall.tools.aggr.MajorityVoteUDAF';
+
+
--------------------------------------------------------------------------------------------------
-- macros available from hive 0.12.0
-- see https://issues.apache.org/jira/browse/HIVE-2655
diff --git a/resources/ddl/define-all.spark b/resources/ddl/define-all.spark
index 89f3def..79381c9 100644
--- a/resources/ddl/define-all.spark
+++ b/resources/ddl/define-all.spark
@@ -839,7 +839,6 @@ sqlContext.sql("CREATE TEMPORARY FUNCTION train_slim AS 'hivemall.recommend.Slim
sqlContext.sql("DROP TEMPORARY FUNCTION IF EXISTS approx_count_distinct")
sqlContext.sql("CREATE TEMPORARY FUNCTION approx_count_distinct AS 'hivemall.sketch.hll.ApproxCountDistinctUDAF'")
-
/**
* Bloom Filter
*/
@@ -862,3 +861,15 @@ sqlContext.sql("CREATE TEMPORARY FUNCTION bloom_or AS 'hivemall.sketch.bloom.Blo
sqlContext.sql("DROP TEMPORARY FUNCTION IF EXISTS bloom_contains_any")
sqlContext.sql("CREATE TEMPORARY FUNCTION bloom_contains_any AS 'hivemall.sketch.bloom.BloomContainsAnyUDF'")
+/**
+ * Aggregation
+ */
+
+sqlContext.sql("DROP TEMPORARY FUNCTION IF EXISTS max_by")
+sqlContext.sql("CREATE TEMPORARY FUNCTION max_by AS 'hivemall.tools.aggr.MaxByUDAF'")
+
+sqlContext.sql("DROP TEMPORARY FUNCTION IF EXISTS min_by")
+sqlContext.sql("CREATE TEMPORARY FUNCTION min_by AS 'hivemall.tools.aggr.MinByUDAF'")
+
+sqlContext.sql("DROP TEMPORARY FUNCTION IF EXISTS majority_vote")
+sqlContext.sql("CREATE TEMPORARY FUNCTION majority_vote AS 'hivemall.tools.aggr.MajorityVoteUDAF'")
diff --git a/tools/hivemall-docs/src/main/java/hivemall/docs/FuncsListGeneratorMojo.java b/tools/hivemall-docs/src/main/java/hivemall/docs/FuncsListGeneratorMojo.java
index 0544e99..3e7bbdb 100644
--- a/tools/hivemall-docs/src/main/java/hivemall/docs/FuncsListGeneratorMojo.java
+++ b/tools/hivemall-docs/src/main/java/hivemall/docs/FuncsListGeneratorMojo.java
@@ -80,6 +80,7 @@ public class FuncsListGeneratorMojo extends AbstractMojo {
private static final Map<String, List<String>> genericFuncsHeaders = new LinkedHashMap<>();
static {
+ genericFuncsHeaders.put("# Aggregation", Arrays.asList("hivemall.tools.aggr"));
genericFuncsHeaders.put("# Array",
Arrays.asList("hivemall.tools.array", "hivemall.tools.list"));
genericFuncsHeaders.put("# Bitset", Collections.singletonList("hivemall.tools.bits"));