[spark] branch master updated: [SPARK-26353][SQL] Add typed aggregate functions(max/min) to the example module.

[gu...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

gurwls223 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git


The following commit(s) were added to refs/heads/master by this push:
     new 8290e5e  [SPARK-26353][SQL] Add typed aggregate functions(max/min) to the example module.
8290e5e is described below

commit 8290e5eccb22e6b865075a0fe62772ce072f0a08
Author: liuxian <li...@zte.com.cn>
AuthorDate: Mon Feb 18 17:20:58 2019 +0800

    [SPARK-26353][SQL] Add typed aggregate functions(max/min) to the example module.
    
    ## What changes were proposed in this pull request?
    
    Add typed aggregate functions(max/min) to the example module.
    
    ## How was this patch tested?
    Manual testing:
    running typed minimum:
    
    ```
    +-----+----------------------+
    |value|TypedMin(scala.Tuple2)|
    +-----+----------------------+
    |    0|                 [0.0]|
    |    2|                 [2.0]|
    |    1|                 [1.0]|
    +-----+----------------------+
    ```
    
    running typed maximum:
    
    ```
    +-----+----------------------+
    |value|TypedMax(scala.Tuple2)|
    +-----+----------------------+
    |    0|                  [18]|
    |    2|                  [17]|
    |    1|                  [19]|
    +-----+----------------------+
    ```
    
    Closes #23304 from 10110346/typedminmax.
    
    Authored-by: liuxian <li...@zte.com.cn>
    Signed-off-by: Hyukjin Kwon <gu...@apache.org>
---
 .../spark/examples/sql/SimpleTypedAggregator.scala | 74 ++++++++++++++++++++++
 1 file changed, 74 insertions(+)

diff --git a/examples/src/main/scala/org/apache/spark/examples/sql/SimpleTypedAggregator.scala b/examples/src/main/scala/org/apache/spark/examples/sql/SimpleTypedAggregator.scala
index f8af919..5510f00 100644
--- a/examples/src/main/scala/org/apache/spark/examples/sql/SimpleTypedAggregator.scala
+++ b/examples/src/main/scala/org/apache/spark/examples/sql/SimpleTypedAggregator.scala
@@ -44,6 +44,12 @@ object SimpleTypedAggregator {
     println("running typed average:")
     ds.groupByKey(_._1).agg(new TypedAverage[(Long, Long)](_._2.toDouble).toColumn).show()
 
+    println("running typed minimum:")
+    ds.groupByKey(_._1).agg(new TypedMin[(Long, Long)](_._2.toDouble).toColumn).show()
+
+    println("running typed maximum:")
+    ds.groupByKey(_._1).agg(new TypedMax[(Long, Long)](_._2).toColumn).show()
+
     spark.stop()
   }
 }
@@ -84,3 +90,71 @@ class TypedAverage[IN](val f: IN => Double) extends Aggregator[IN, (Double, Long
   }
   override def outputEncoder: Encoder[Double] = Encoders.scalaDouble
 }
+
+class TypedMin[IN](val f: IN => Double) extends Aggregator[IN, MutableDouble, Option[Double]] {
+  override def zero: MutableDouble = null
+  override def reduce(b: MutableDouble, a: IN): MutableDouble = {
+    if (b == null) {
+      new MutableDouble(f(a))
+    } else {
+      b.value = math.min(b.value, f(a))
+      b
+    }
+  }
+  override def merge(b1: MutableDouble, b2: MutableDouble): MutableDouble = {
+    if (b1 == null) {
+      b2
+    } else if (b2 == null) {
+      b1
+    } else {
+      b1.value = math.min(b1.value, b2.value)
+      b1
+    }
+  }
+  override def finish(reduction: MutableDouble): Option[Double] = {
+    if (reduction != null) {
+      Some(reduction.value)
+    } else {
+      None
+    }
+  }
+
+  override def bufferEncoder: Encoder[MutableDouble] = Encoders.kryo[MutableDouble]
+  override def outputEncoder: Encoder[Option[Double]] = Encoders.product[Option[Double]]
+}
+
+class TypedMax[IN](val f: IN => Long) extends Aggregator[IN, MutableLong, Option[Long]] {
+  override def zero: MutableLong = null
+  override def reduce(b: MutableLong, a: IN): MutableLong = {
+    if (b == null) {
+      new MutableLong(f(a))
+    } else {
+      b.value = math.max(b.value, f(a))
+      b
+    }
+  }
+  override def merge(b1: MutableLong, b2: MutableLong): MutableLong = {
+    if (b1 == null) {
+      b2
+    } else if (b2 == null) {
+      b1
+    } else {
+      b1.value = math.max(b1.value, b2.value)
+      b1
+    }
+  }
+  override def finish(reduction: MutableLong): Option[Long] = {
+    if (reduction != null) {
+      Some(reduction.value)
+    } else {
+      None
+    }
+  }
+
+  override def bufferEncoder: Encoder[MutableLong] = Encoders.kryo[MutableLong]
+  override def outputEncoder: Encoder[Option[Long]] = Encoders.product[Option[Long]]
+}
+
+class MutableLong(var value: Long) extends Serializable
+
+class MutableDouble(var value: Double) extends Serializable


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org