[SPARK-26353][SQL]Add typed aggregate functions(max/min) to the example module. #23304

10110346 · 2018-12-13T01:23:32Z

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]|
+-----+----------------------+

SparkQA · 2018-12-13T04:53:07Z

Test build #100056 has finished for PR 23304 at commit defee03.

This patch fails SparkR unit tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
class TypedMaxDouble[IN](val f: IN => Double) extends Aggregator[IN, Double, Double]
class TypedMaxLong[IN](val f: IN => Long) extends Aggregator[IN, Long, Long]
class TypedMinDouble[IN](val f: IN => Double) extends Aggregator[IN, Double, Double]
class TypedMinLong[IN](val f: IN => Long) extends Aggregator[IN, Long, Long]

10110346 · 2018-12-13T06:06:04Z

retest this please

SparkQA · 2018-12-13T08:05:02Z

Test build #100071 has finished for PR 23304 at commit defee03.

This patch fails due to an unknown error code, -9.
This patch merges cleanly.
This patch adds the following public classes (experimental):
class TypedMaxDouble[IN](val f: IN => Double) extends Aggregator[IN, Double, Double]
class TypedMaxLong[IN](val f: IN => Long) extends Aggregator[IN, Long, Long]
class TypedMinDouble[IN](val f: IN => Double) extends Aggregator[IN, Double, Double]
class TypedMinLong[IN](val f: IN => Long) extends Aggregator[IN, Long, Long]

10110346 · 2018-12-13T10:40:57Z

retest this please

SparkQA · 2018-12-13T14:25:43Z

Test build #100084 has finished for PR 23304 at commit defee03.

This patch passes all tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
class TypedMaxDouble[IN](val f: IN => Double) extends Aggregator[IN, Double, Double]
class TypedMaxLong[IN](val f: IN => Long) extends Aggregator[IN, Long, Long]
class TypedMinDouble[IN](val f: IN => Double) extends Aggregator[IN, Double, Double]
class TypedMinLong[IN](val f: IN => Long) extends Aggregator[IN, Long, Long]

10110346 · 2018-12-14T03:15:10Z

cc @rxin @cloud-fan

gatorsmile · 2018-12-14T05:29:34Z

See the PR #18113

10110346 · 2018-12-14T06:32:59Z

This PR #18113 has a better solution, thanks @gatorsmile
But it looks like it hasn't been maintained.

10110346 · 2018-12-14T08:21:03Z

This PR #18113 doesn't look like making progress for a long time so far, I take this PR over.

SparkQA · 2018-12-14T11:44:05Z

Test build #100136 has finished for PR 23304 at commit defee03.

This patch passes all tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
class TypedMaxDouble[IN](val f: IN => Double) extends Aggregator[IN, Double, Double]
class TypedMaxLong[IN](val f: IN => Long) extends Aggregator[IN, Long, Long]
class TypedMinDouble[IN](val f: IN => Double) extends Aggregator[IN, Double, Double]
class TypedMinLong[IN](val f: IN => Long) extends Aggregator[IN, Long, Long]

SparkQA · 2018-12-14T12:02:47Z

Test build #100139 has finished for PR 23304 at commit 0b64ae7.

This patch passes all tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
trait TypedMinDouble[IN, OUT] extends Aggregator[IN, MutableDouble, OUT]
class JavaTypedMinDouble[IN](override val f: IN => Double)
class ScalaTypedMinDouble[IN](override val f: IN => Double)
trait TypedMaxDouble[IN, OUT] extends Aggregator[IN, MutableDouble, OUT]
class JavaTypedMaxDouble[IN](override val f: IN => Double)
class ScalaTypedMaxDouble[IN](override val f: IN => Double)
trait TypedMinLong[IN, OUT] extends Aggregator[IN, MutableLong, OUT]
class JavaTypedMinLong[IN](override val f: IN => Long) extends TypedMinLong[IN, java.lang.Long]
class ScalaTypedMinLong[IN](override val f: IN => Long) extends TypedMinLong[IN, Option[Long]]
trait TypedMaxLong[IN, OUT] extends Aggregator[IN, MutableLong, OUT]
class JavaTypedMaxLong[IN](override val f: IN => Long) extends TypedMaxLong[IN, java.lang.Long]
class ScalaTypedMaxLong[IN](override val f: IN => Long) extends TypedMaxLong[IN, Option[Long]]
class MutableLong(var value: Long) extends Serializable
class MutableDouble(var value: Double) extends Serializable

cloud-fan · 2018-12-16T01:53:37Z

do you resolve #18113 (comment) ?

10110346 · 2018-12-17T01:40:29Z

@cloud-fan No, I don't

HyukjinKwon · 2019-01-06T04:24:44Z

?? then are you going to resolve that comment?

10110346 · 2019-01-07T06:21:29Z

?? then are you going to resolve that comment?

I can't solve this problem at present, someone can help me?

cloud-fan · 2019-01-07T11:05:15Z

How did you fix the tests? #18113 couldn't pass tests because of that problem.

10110346 · 2019-01-07T12:28:53Z

How did you fix the tests? #18113 couldn't pass tests because of that problem.

I just duplicated his code, I didn't do anything else.

cloud-fan · 2019-01-07T14:33:31Z

sql/core/src/main/java/org/apache/spark/sql/expressions/javalang/typed.java

+  /**
+   * Min aggregate function for floating point (double) type.
+   *
+   * @since 2.4.0


it's 3.0.0 now

cloud-fan · 2019-01-07T14:40:21Z

sql/core/src/main/scala/org/apache/spark/sql/execution/aggregate/typedaggregators.scala

+class JavaTypedMinDouble[IN](override val f: IN => Double)
+  extends TypedMinDouble[IN, java.lang.Double] {
+  override def outputEncoder: Encoder[java.lang.Double] = ExpressionEncoder[java.lang.Double]()
+  override def finish(reduction: MutableDouble): java.lang.Double = reduction.value


what if reduction is null?

thanks, I will update

cloud-fan · 2019-01-07T14:41:01Z

sql/core/src/main/scala/org/apache/spark/sql/execution/aggregate/typedaggregators.scala

+class JavaTypedMaxDouble[IN](override val f: IN => Double)
+  extends TypedMaxDouble[IN, java.lang.Double] {
+  override def outputEncoder: Encoder[java.lang.Double] = ExpressionEncoder[java.lang.Double]()
+  override def finish(reduction: MutableDouble): java.lang.Double = reduction.value


cloud-fan · 2019-01-07T14:41:31Z

sql/core/src/main/scala/org/apache/spark/sql/execution/aggregate/typedaggregators.scala

+
+class JavaTypedMinLong[IN](override val f: IN => Long) extends TypedMinLong[IN, java.lang.Long] {
+  override def outputEncoder: Encoder[java.lang.Long] = ExpressionEncoder[java.lang.Long]()
+  override def finish(reduction: MutableLong): java.lang.Long = reduction.value


cloud-fan · 2019-01-07T14:41:53Z

sql/core/src/main/scala/org/apache/spark/sql/execution/aggregate/typedaggregators.scala

+
+class JavaTypedMaxLong[IN](override val f: IN => Long) extends TypedMaxLong[IN, java.lang.Long] {
+  override def outputEncoder: Encoder[java.lang.Long] = ExpressionEncoder[java.lang.Long]()
+  override def finish(reduction: MutableLong): java.lang.Long = reduction.value


cloud-fan · 2019-01-07T14:48:43Z

sql/core/src/test/java/test/org/apache/spark/sql/Java8DatasetAggregatorSuite.java

+    KeyValueGroupedDataset<String, Tuple2<String, Integer>> grouped = generateGroupedDataset();
+    Dataset<Tuple2<String, Double>> agged = grouped.agg(typed.min(v -> (double)v._2()));
+    Assert.assertEquals(
+            Arrays.asList(new Tuple2<>("a", 1.0), new Tuple2<>("b", 3.0)),


4 space indentation

thanks, I will update

cloud-fan · 2019-01-07T14:52:19Z

sql/core/src/test/scala/org/apache/spark/sql/DatasetAggregatorSuite.scala

+      ("b", Some(-4.0), Some(-4L), Some(4.0), Some(4L)))
+  }
+
+  test("typed aggregate: empty") {


can we test empty input with global aggregate in the java test as well?

SparkQA · 2019-01-14T08:05:02Z

Test build #101163 has finished for PR 23304 at commit ae2152b.

This patch fails due to an unknown error code, -9.
This patch merges cleanly.
This patch adds the following public classes (experimental):
trait TypedMinDouble[IN, OUT] extends Aggregator[IN, MutableDouble, OUT]
class JavaTypedMinDouble[IN](override val f: IN => Double)
class ScalaTypedMinDouble[IN](override val f: IN => Double)
trait TypedMaxDouble[IN, OUT] extends Aggregator[IN, MutableDouble, OUT]
class JavaTypedMaxDouble[IN](override val f: IN => Double)
class ScalaTypedMaxDouble[IN](override val f: IN => Double)
trait TypedMinLong[IN, OUT] extends Aggregator[IN, MutableLong, OUT]
class JavaTypedMinLong[IN](override val f: IN => Long) extends TypedMinLong[IN, java.lang.Long]
class ScalaTypedMinLong[IN](override val f: IN => Long) extends TypedMinLong[IN, Option[Long]]
trait TypedMaxLong[IN, OUT] extends Aggregator[IN, MutableLong, OUT]
class JavaTypedMaxLong[IN](override val f: IN => Long) extends TypedMaxLong[IN, java.lang.Long]
class ScalaTypedMaxLong[IN](override val f: IN => Long) extends TypedMaxLong[IN, Option[Long]]
class MutableLong(var value: Long) extends Serializable
class MutableDouble(var value: Double) extends Serializable

10110346 · 2019-01-14T08:23:41Z

retest this please

SparkQA · 2019-01-14T12:16:07Z

Test build #101170 has finished for PR 23304 at commit ae2152b.

This patch passes all tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
trait TypedMinDouble[IN, OUT] extends Aggregator[IN, MutableDouble, OUT]
class JavaTypedMinDouble[IN](override val f: IN => Double)
class ScalaTypedMinDouble[IN](override val f: IN => Double)
trait TypedMaxDouble[IN, OUT] extends Aggregator[IN, MutableDouble, OUT]
class JavaTypedMaxDouble[IN](override val f: IN => Double)
class ScalaTypedMaxDouble[IN](override val f: IN => Double)
trait TypedMinLong[IN, OUT] extends Aggregator[IN, MutableLong, OUT]
class JavaTypedMinLong[IN](override val f: IN => Long) extends TypedMinLong[IN, java.lang.Long]
class ScalaTypedMinLong[IN](override val f: IN => Long) extends TypedMinLong[IN, Option[Long]]
trait TypedMaxLong[IN, OUT] extends Aggregator[IN, MutableLong, OUT]
class JavaTypedMaxLong[IN](override val f: IN => Long) extends TypedMaxLong[IN, java.lang.Long]
class ScalaTypedMaxLong[IN](override val f: IN => Long) extends TypedMaxLong[IN, Option[Long]]
class MutableLong(var value: Long) extends Serializable
class MutableDouble(var value: Double) extends Serializable

HyukjinKwon · 2019-01-14T12:30:58Z

sql/core/src/main/java/org/apache/spark/sql/expressions/javalang/typed.java

+  }
+
+  /**
+   * Min aggregate function for floating point (double) type.


Min -> Max. Actually I will elaborate it more. Please avoid abbreviation like min or max in doc.

HyukjinKwon · 2019-01-14T12:33:56Z

Looks fine to me too.

SparkQA · 2019-01-15T02:39:30Z

Test build #101212 has finished for PR 23304 at commit 04c0a22.

This patch fails Spark unit tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
trait TypedMinDouble[IN, OUT] extends Aggregator[IN, MutableDouble, OUT]
class JavaTypedMinDouble[IN](override val f: IN => Double)
class ScalaTypedMinDouble[IN](override val f: IN => Double)
trait TypedMaxDouble[IN, OUT] extends Aggregator[IN, MutableDouble, OUT]
class JavaTypedMaxDouble[IN](override val f: IN => Double)
class ScalaTypedMaxDouble[IN](override val f: IN => Double)
trait TypedMinLong[IN, OUT] extends Aggregator[IN, MutableLong, OUT]
class JavaTypedMinLong[IN](override val f: IN => Long) extends TypedMinLong[IN, java.lang.Long]
class ScalaTypedMinLong[IN](override val f: IN => Long) extends TypedMinLong[IN, Option[Long]]
trait TypedMaxLong[IN, OUT] extends Aggregator[IN, MutableLong, OUT]
class JavaTypedMaxLong[IN](override val f: IN => Long) extends TypedMaxLong[IN, java.lang.Long]
class ScalaTypedMaxLong[IN](override val f: IN => Long) extends TypedMaxLong[IN, Option[Long]]
class MutableLong(var value: Long) extends Serializable
class MutableDouble(var value: Double) extends Serializable

HyukjinKwon · 2019-01-15T02:44:09Z

retest this please

SparkQA · 2019-01-15T06:55:14Z

Test build #101219 has finished for PR 23304 at commit 04c0a22.

This patch passes all tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
trait TypedMinDouble[IN, OUT] extends Aggregator[IN, MutableDouble, OUT]
class JavaTypedMinDouble[IN](override val f: IN => Double)
class ScalaTypedMinDouble[IN](override val f: IN => Double)
trait TypedMaxDouble[IN, OUT] extends Aggregator[IN, MutableDouble, OUT]
class JavaTypedMaxDouble[IN](override val f: IN => Double)
class ScalaTypedMaxDouble[IN](override val f: IN => Double)
trait TypedMinLong[IN, OUT] extends Aggregator[IN, MutableLong, OUT]
class JavaTypedMinLong[IN](override val f: IN => Long) extends TypedMinLong[IN, java.lang.Long]
class ScalaTypedMinLong[IN](override val f: IN => Long) extends TypedMinLong[IN, Option[Long]]
trait TypedMaxLong[IN, OUT] extends Aggregator[IN, MutableLong, OUT]
class JavaTypedMaxLong[IN](override val f: IN => Long) extends TypedMaxLong[IN, java.lang.Long]
class ScalaTypedMaxLong[IN](override val f: IN => Long) extends TypedMaxLong[IN, Option[Long]]
class MutableLong(var value: Long) extends Serializable
class MutableDouble(var value: Double) extends Serializable

rxin · 2019-01-15T07:48:51Z

I'm kind of wondering whether it'd make sense to add these. It adds a lot of code which would incur some maintenance cost, and users can trivially implement these themselves, or just use the untyped version, and we don't need to spend time discussing whether these should follow SQL semantics or Scala semantics.

HyukjinKwon · 2019-01-15T10:41:22Z

Actually, WDYT about ..

Deprecating or removing them? Looks they are experimental.
Stopping adding this from now on (but get this in as the last case considering min/max/count are pretty common).

Was having the kind of similar impression.

HeartSaVioR · 2019-02-09T00:14:27Z

IMHO it would be ideal to have common aggregate functions for typed version as well (agree to restrict them to like sum/avg/count/min/max), otherwise end users using typed API have to implement such basic thing on their own.

Btw, TODOs in source code are tend to being considered as things to contribute. Might be ideal to remove these false alarm once we don't see its needs.

cloud-fan · 2019-02-15T01:08:45Z

Hi @10110346 , can you move these new functions to the "example" package?

10110346 · 2019-02-15T02:22:14Z

Hi @10110346 , can you move these new functions to the "example" package?

Ok, I will do it, thanks

SparkQA · 2019-02-15T08:59:02Z

Test build #102382 has finished for PR 23304 at commit bbb820b.

This patch passes all tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
class TypedMin[IN](val f: IN => Double) extends Aggregator[IN, MutableDouble, Option[Double]]
class TypedMax[IN](val f: IN => Long) extends Aggregator[IN, MutableLong, Option[Long]]
class MutableLong(var value: Long) extends Serializable
class MutableDouble(var value: Double) extends Serializable

HyukjinKwon · 2019-02-15T09:07:01Z

examples/src/main/scala/org/apache/spark/examples/sql/SimpleTypedAggregator.scala

@@ -44,6 +45,12 @@ object SimpleTypedAggregator {
    println("running typed average:")
    ds.groupByKey(_._1).agg(new TypedAverage[(Long, Long)](_._2.toDouble).toColumn).show()

+    println("running typed Minimum:")


Minimum -> minimum to be consistent

SparkQA · 2019-02-15T09:37:53Z

Test build #102385 has finished for PR 23304 at commit 20aef3e.

This patch passes all tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
class TypedMin[IN](val f: IN => Double) extends Aggregator[IN, MutableDouble, Option[Double]]
class TypedMax[IN](val f: IN => Long) extends Aggregator[IN, MutableLong, Option[Long]]
class MutableLong(var value: Long) extends Serializable
class MutableDouble(var value: Double) extends Serializable

cloud-fan · 2019-02-15T15:13:06Z

examples/src/main/scala/org/apache/spark/examples/sql/SimpleTypedAggregator.scala

+  }
+
+  override def bufferEncoder: Encoder[MutableDouble] = Encoders.kryo[MutableDouble]
+  override def outputEncoder: Encoder[Option[Double]] = ExpressionEncoder[Option[Double]]()


can we use Encoders.product[Option[Double]]? Or we can use Encoders.DOUBLE and use null to represent None.

The thing is, ExpressionEncoder is an internal class and we should avoid exposing it in the example module.

SparkQA · 2019-02-18T01:57:43Z

Test build #102440 has finished for PR 23304 at commit 26a448d.

This patch passes all tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
class TypedMin[IN](val f: IN => Double) extends Aggregator[IN, MutableDouble, Option[Double]]
class TypedMax[IN](val f: IN => Long) extends Aggregator[IN, MutableLong, Option[Long]]
class MutableLong(var value: Long) extends Serializable
class MutableDouble(var value: Double) extends Serializable

cloud-fan · 2019-02-18T04:09:52Z

thanks, merging to master!

10110346 · 2019-02-18T07:58:52Z

@cloud-fan It seems merging has not succeeded, thanks

HyukjinKwon · 2019-02-18T09:18:48Z

Merged to master

10110346 changed the title ~~[SPARK-26353][SQL]Add typed aggregate functions:max&&min~~ [SPARK-26353][SQL]Add typed aggregate functions: max&&min Dec 13, 2018

10110346 closed this Dec 14, 2018

HeartSaVioR mentioned this pull request Dec 14, 2018

[SPARK-20890][SQL] Added min and max typed aggregation functions #18113

Closed

10110346 reopened this Dec 14, 2018

10110346 force-pushed the typedminmax branch from defee03 to 0b64ae7 Compare December 14, 2018 08:16

cloud-fan reviewed Jan 7, 2019

View reviewed changes

10110346 force-pushed the typedminmax branch from 0b64ae7 to eafc50b Compare January 9, 2019 10:04

HyukjinKwon reviewed Jan 14, 2019

View reviewed changes

10110346 force-pushed the typedminmax branch from ae2152b to 04c0a22 Compare January 15, 2019 01:01

10110346 force-pushed the typedminmax branch from 04c0a22 to bbb820b Compare February 15, 2019 08:42

10110346 changed the title ~~[SPARK-26353][SQL]Add typed aggregate functions: max&&min~~ [SPARK-26353][SQL]Add typed aggregate functions(max/min) to the example module. Feb 15, 2019

HyukjinKwon reviewed Feb 15, 2019

View reviewed changes

HyukjinKwon approved these changes Feb 15, 2019

View reviewed changes

10110346 force-pushed the typedminmax branch from bbb820b to 20aef3e Compare February 15, 2019 09:23

cloud-fan reviewed Feb 15, 2019

View reviewed changes

Add typed aggregate functions:max&&min

26a448d

10110346 force-pushed the typedminmax branch from 20aef3e to 26a448d Compare February 18, 2019 01:41

HyukjinKwon closed this in 8290e5e Feb 18, 2019

[SPARK-26353][SQL]Add typed aggregate functions(max/min) to the example module. #23304

[SPARK-26353][SQL]Add typed aggregate functions(max/min) to the example module. #23304

Conversation

10110346 commented Dec 13, 2018 • edited by HyukjinKwon

What changes were proposed in this pull request?

How was this patch tested?

SparkQA commented Dec 13, 2018

10110346 commented Dec 13, 2018

SparkQA commented Dec 13, 2018

10110346 commented Dec 13, 2018

SparkQA commented Dec 13, 2018

10110346 commented Dec 14, 2018

gatorsmile commented Dec 14, 2018

10110346 commented Dec 14, 2018

10110346 commented Dec 14, 2018

SparkQA commented Dec 14, 2018

SparkQA commented Dec 14, 2018

cloud-fan commented Dec 16, 2018

10110346 commented Dec 17, 2018

HyukjinKwon commented Jan 6, 2019

10110346 commented Jan 7, 2019

cloud-fan commented Jan 7, 2019

10110346 commented Jan 7, 2019

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

SparkQA commented Jan 14, 2019

10110346 commented Jan 14, 2019

SparkQA commented Jan 14, 2019

Choose a reason for hiding this comment

Choose a reason for hiding this comment

HyukjinKwon commented Jan 14, 2019

SparkQA commented Jan 15, 2019

HyukjinKwon commented Jan 15, 2019

SparkQA commented Jan 15, 2019

rxin commented Jan 15, 2019

HyukjinKwon commented Jan 15, 2019

HeartSaVioR commented Feb 9, 2019

cloud-fan commented Feb 15, 2019

10110346 commented Feb 15, 2019

SparkQA commented Feb 15, 2019

Choose a reason for hiding this comment

Choose a reason for hiding this comment

SparkQA commented Feb 15, 2019

Choose a reason for hiding this comment

Choose a reason for hiding this comment

SparkQA commented Feb 18, 2019

cloud-fan commented Feb 18, 2019

10110346 commented Feb 18, 2019

HyukjinKwon commented Feb 18, 2019

10110346 commented Dec 13, 2018 •

edited by HyukjinKwon