Re: [PR] [SPARK-48008][WIP] Support UDAFs in Spark Connect [spark]
xupefei commented on code in PR #46245: URL: https://github.com/apache/spark/pull/46245#discussion_r1610072254 ## connector/connect/client/jvm/src/main/scala/org/apache/spark/sql/expressions/UserDefinedFunction.scala: ## @@ -103,7 +103,8 @@ case class ScalarUserDefinedFunction private[sql] ( outputType: proto.DataType, name: Option[String], override val nullable: Boolean, -override val deterministic: Boolean) +override val deterministic: Boolean, +aggregate: Boolean) Review Comment: I can't - there is another `UserDefinedFunction` defined in Spark Core. I renamed this one to `ScalaUserDefinedFunction` instead. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org - To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org For additional commands, e-mail: reviews-h...@spark.apache.org
Re: [PR] [SPARK-48008][WIP] Support UDAFs in Spark Connect [spark]
hvanhovell commented on code in PR #46245:
URL: https://github.com/apache/spark/pull/46245#discussion_r1608851090
##
connector/connect/client/jvm/src/test/scala/org/apache/spark/sql/UserDefinedFunctionE2ETestSuite.scala:
##
@@ -346,4 +347,42 @@ class UserDefinedFunctionE2ETestSuite extends QueryTest {
val result = df.select(f($"id")).as[Long].head()
assert(result == 1L)
}
+
+ test("UDAF custom Aggregator - primitive types") {
Review Comment:
Can you add a test for a UDAF with a custom toColumn implementation?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
Re: [PR] [SPARK-48008][WIP] Support UDAFs in Spark Connect [spark]
hvanhovell commented on code in PR #46245: URL: https://github.com/apache/spark/pull/46245#discussion_r1608832100 ## connector/connect/client/jvm/src/main/scala/org/apache/spark/sql/expressions/UserDefinedFunction.scala: ## @@ -103,7 +103,8 @@ case class ScalarUserDefinedFunction private[sql] ( outputType: proto.DataType, name: Option[String], override val nullable: Boolean, -override val deterministic: Boolean) +override val deterministic: Boolean, +aggregate: Boolean) Review Comment: Can you rename the class to `UserDefinedFunction`? -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org - To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org For additional commands, e-mail: reviews-h...@spark.apache.org
Re: [PR] [SPARK-48008][WIP] Support UDAFs in Spark Connect [spark]
hvanhovell commented on code in PR #46245:
URL: https://github.com/apache/spark/pull/46245#discussion_r1608831659
##
connector/connect/client/jvm/src/main/scala/org/apache/spark/sql/expressions/Aggregator.scala:
##
@@ -0,0 +1,104 @@
+/*
+ * 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.spark.sql.expressions
+
+import org.apache.spark.sql.{Encoder, TypedColumn}
+
+/**
+ * A base class for user-defined aggregations, which can be used in `Dataset`
operations to take
+ * all of the elements of a group and reduce them to a single value.
+ *
+ * For example, the following aggregator extracts an `int` from a specific
class and adds them up:
+ * {{{
+ * case class Data(i: Int)
+ *
+ * val customSummer = new Aggregator[Data, Int, Int] {
+ * def zero: Int = 0
+ * def reduce(b: Int, a: Data): Int = b + a.i
+ * def merge(b1: Int, b2: Int): Int = b1 + b2
+ * def finish(r: Int): Int = r
+ * def bufferEncoder: Encoder[Int] = Encoders.scalaInt
+ * def outputEncoder: Encoder[Int] = Encoders.scalaInt
+ * }
+ *
+ * spark.udf.register("customSummer", udaf(customSummer))
+ * val ds: Dataset[Data] = ...
+ * val aggregated = ds.selectExpr("customSummer(i)")
+ * }}}
+ *
+ * Based loosely on Aggregator from Algebird:
https://github.com/twitter/algebird
+ *
+ * @tparam IN The input type for the aggregation.
+ * @tparam BUF The type of the intermediate value of the reduction.
+ * @tparam OUT The type of the final output result.
+ * @since 4.0.0
+ */
+@SerialVersionUID(2093413866369130093L)
+abstract class Aggregator[-IN, BUF, OUT] extends Serializable {
+
+ /**
+ * A zero value for this aggregation. Should satisfy the property that any b
+ zero = b.
+ *
+ * @since 4.0.0
+ */
+ def zero: BUF
+
+ /**
+ * Combine two values to produce a new value. For performance, the function
may modify `b` and
+ * return it instead of constructing new object for b.
+ *
+ * @since 4.0.0
+ */
+ def reduce(b: BUF, a: IN): BUF
+
+ /**
+ * Merge two intermediate values.
+ *
+ * @since 4.0.0
+ */
+ def merge(b1: BUF, b2: BUF): BUF
+
+ /**
+ * Transform the output of the reduction.
+ *
+ * @since 4.0.0
+ */
+ def finish(reduction: BUF): OUT
+
+ /**
+ * Specifies the `Encoder` for the intermediate value type.
+ *
+ * @since 4.0.0
+ */
+ def bufferEncoder: Encoder[BUF]
+
+ /**
+ * Specifies the `Encoder` for the final output value type.
+ *
+ * @since 4.0.0
+ */
+ def outputEncoder: Encoder[OUT]
+
+ /**
+ * Returns this `Aggregator` as a `TypedColumn` that can be used in
`Dataset`.
+ * operations.
+ */
+ def toColumn: TypedColumn[IN, OUT] = {
Review Comment:
We need a test where we override toColumn()...
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
Re: [PR] [SPARK-48008][WIP] Support UDAFs in Spark Connect [spark]
hvanhovell commented on code in PR #46245:
URL: https://github.com/apache/spark/pull/46245#discussion_r1608830698
##
connector/connect/client/jvm/src/main/scala/org/apache/spark/sql/expressions/Aggregator.scala:
##
@@ -0,0 +1,104 @@
+/*
+ * 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.spark.sql.expressions
+
+import org.apache.spark.sql.{Encoder, TypedColumn}
+
+/**
+ * A base class for user-defined aggregations, which can be used in `Dataset`
operations to take
+ * all of the elements of a group and reduce them to a single value.
+ *
+ * For example, the following aggregator extracts an `int` from a specific
class and adds them up:
+ * {{{
+ * case class Data(i: Int)
+ *
+ * val customSummer = new Aggregator[Data, Int, Int] {
+ * def zero: Int = 0
+ * def reduce(b: Int, a: Data): Int = b + a.i
+ * def merge(b1: Int, b2: Int): Int = b1 + b2
+ * def finish(r: Int): Int = r
+ * def bufferEncoder: Encoder[Int] = Encoders.scalaInt
+ * def outputEncoder: Encoder[Int] = Encoders.scalaInt
+ * }
+ *
+ * spark.udf.register("customSummer", udaf(customSummer))
+ * val ds: Dataset[Data] = ...
+ * val aggregated = ds.selectExpr("customSummer(i)")
+ * }}}
+ *
+ * Based loosely on Aggregator from Algebird:
https://github.com/twitter/algebird
+ *
+ * @tparam IN The input type for the aggregation.
+ * @tparam BUF The type of the intermediate value of the reduction.
+ * @tparam OUT The type of the final output result.
+ * @since 4.0.0
+ */
+@SerialVersionUID(2093413866369130093L)
+abstract class Aggregator[-IN, BUF, OUT] extends Serializable {
Review Comment:
I don't think we can move it. Column currently has two implementations. It
does raise a question if it'll work
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
Re: [PR] [SPARK-48008][WIP] Support UDAFs in Spark Connect [spark]
xupefei commented on code in PR #46245:
URL: https://github.com/apache/spark/pull/46245#discussion_r1596852526
##
sql/core/src/main/scala/org/apache/spark/sql/expressions/Aggregator.scala:
##
@@ -49,6 +49,7 @@ import
org.apache.spark.sql.execution.aggregate.TypedAggregateExpression
* @tparam OUT The type of the final output result.
* @since 1.6.0
*/
+@SerialVersionUID(2093413866369130093L)
Review Comment:
I tried multiple methods and still can't succeed without this UID. As long
as we want users to define UDAF like this:
```
new Aggregator {
def merge(...) = {...}
...
}
```
The serialized payload (either the whole Aggregator instance or individual
methods (e.g., `agg.merge _`)) will carry a reference to the Aggregator
instance that needs to be decoded on the server side. Without this UID, the
decode will fail with an error message `class UID not match`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
Re: [PR] [SPARK-48008][WIP] Support UDAFs in Spark Connect [spark]
xupefei commented on code in PR #46245:
URL: https://github.com/apache/spark/pull/46245#discussion_r1601524741
##
connector/connect/common/src/main/protobuf/spark/connect/expressions.proto:
##
@@ -379,6 +380,15 @@ message ScalarScalaUDF {
bool nullable = 4;
}
+message ScalaUDAF {
+ // (Required) Serialized JVM object of the Aggregator instance, including
buffer and output encoders
+ bytes payload = 1;
+ // (Required) Input type of the UDAF
+ DataType inputType = 2;
+ // (Required) True if the UDAF can return null value
+ bool nullable = 3;
Review Comment:
Added back!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
Re: [PR] [SPARK-48008][WIP] Support UDAFs in Spark Connect [spark]
xupefei commented on code in PR #46245:
URL: https://github.com/apache/spark/pull/46245#discussion_r1601524509
##
connector/connect/client/jvm/src/main/scala/org/apache/spark/sql/expressions/UserDefinedFunction.scala:
##
@@ -205,6 +205,91 @@ object ScalarUserDefinedFunction {
}
}
+case class UserDefinedAggregationFunction private[sql] (
Review Comment:
A lot. I just did a refactoring to re-use the ScalarUserDefinedFunction
proto and most of the Scala code.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
Re: [PR] [SPARK-48008][WIP] Support UDAFs in Spark Connect [spark]
xupefei commented on code in PR #46245:
URL: https://github.com/apache/spark/pull/46245#discussion_r1596852526
##
sql/core/src/main/scala/org/apache/spark/sql/expressions/Aggregator.scala:
##
@@ -49,6 +49,7 @@ import
org.apache.spark.sql.execution.aggregate.TypedAggregateExpression
* @tparam OUT The type of the final output result.
* @since 1.6.0
*/
+@SerialVersionUID(2093413866369130093L)
Review Comment:
I tried multiple methods and still can't succeed without this UID. As long
as we want users to use UDAF like this:
```
new Aggregator {
def merge(...) = {...}
...
}
```
The serialized payload (either the whole Aggregator instance or individual
methods) will carry a reference to the Aggregator instance that needs to be
decoded on the server side. Without this UID, it will throw an exception. So
far the only way I found to break this link is to declare interface methods as
high-order functions:
```
new Aggregator {
def merge = (b1, b2) => {...}
...
}
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
Re: [PR] [SPARK-48008][WIP] Support UDAFs in Spark Connect [spark]
xupefei commented on code in PR #46245:
URL: https://github.com/apache/spark/pull/46245#discussion_r1596852526
##
sql/core/src/main/scala/org/apache/spark/sql/expressions/Aggregator.scala:
##
@@ -49,6 +49,7 @@ import
org.apache.spark.sql.execution.aggregate.TypedAggregateExpression
* @tparam OUT The type of the final output result.
* @since 1.6.0
*/
+@SerialVersionUID(2093413866369130093L)
Review Comment:
I tried multiple methods and still can't succeed without this UID. As long
as we want users to use UDAF like this:
```
new Aggregator {
def merge(...) = {...}
...
}
```
The serialized payload (either the whole Aggregator instance or individual
methods) will carry a reference to the Aggregator instance that needs to be
decoded on the server side. Without this UID, it will throw an exception. So
far the only way I found to break this link is to declare interface methods as
high-order functions:
```
new Aggregator {
def merge(...) = (b1, b2) => {...}
...
}
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
Re: [PR] [SPARK-48008][WIP] Support UDAFs in Spark Connect [spark]
xupefei commented on code in PR #46245:
URL: https://github.com/apache/spark/pull/46245#discussion_r1596852526
##
sql/core/src/main/scala/org/apache/spark/sql/expressions/Aggregator.scala:
##
@@ -49,6 +49,7 @@ import
org.apache.spark.sql.execution.aggregate.TypedAggregateExpression
* @tparam OUT The type of the final output result.
* @since 1.6.0
*/
+@SerialVersionUID(2093413866369130093L)
Review Comment:
I tried multiple methods and still can't succeed without this UID. As long
as we want users to use UDAF like this:
```
new Aggregator {
def merge(...) = {...}
...
}
```
The serialized payload (either the whole Aggregator instance or individual
methods) will carry a reference to the Aggregator instance that needs to be
decoded on the server side. Without this UID, it will throw an exception. SO
far the only way to break this link is to declare interface methods as
high-order functions:
```
new Aggregator {
def merge(...) = (b1, b2) => {...}
...
}
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
Re: [PR] [SPARK-48008][WIP] Support UDAFs in Spark Connect [spark]
xupefei commented on code in PR #46245:
URL: https://github.com/apache/spark/pull/46245#discussion_r1596852526
##
sql/core/src/main/scala/org/apache/spark/sql/expressions/Aggregator.scala:
##
@@ -49,6 +49,7 @@ import
org.apache.spark.sql.execution.aggregate.TypedAggregateExpression
* @tparam OUT The type of the final output result.
* @since 1.6.0
*/
+@SerialVersionUID(2093413866369130093L)
Review Comment:
I tried multiple methods and still can't succeed without this UID. As long
as we want users to use UDAF like this:
```
new Aggregator {
def merge(...) = {...}
...
}
```
The serialized payload (either the whole Aggregator instance or individual
methods) will carry a reference to the Aggregator instance that needs to be
decoded on the server side. Without this UID, it will throw an exception.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
Re: [PR] [SPARK-48008][WIP] Support UDAFs in Spark Connect [spark]
xupefei commented on code in PR #46245: URL: https://github.com/apache/spark/pull/46245#discussion_r1596845911 ## sql/core/src/main/scala/org/apache/spark/sql/expressions/Aggregator.scala: ## @@ -49,6 +49,7 @@ import org.apache.spark.sql.execution.aggregate.TypedAggregateExpression * @tparam OUT The type of the final output result. * @since 1.6.0 */ +@SerialVersionUID(2093413866369130093L) Review Comment: It's for mapping the client's `Aggregator` class to this one. This is required because we now serialise the whole `Aggregator` instance on the client side. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org - To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org For additional commands, e-mail: reviews-h...@spark.apache.org
Re: [PR] [SPARK-48008][WIP] Support UDAFs in Spark Connect [spark]
xupefei commented on code in PR #46245: URL: https://github.com/apache/spark/pull/46245#discussion_r1596845911 ## sql/core/src/main/scala/org/apache/spark/sql/expressions/Aggregator.scala: ## @@ -49,6 +49,7 @@ import org.apache.spark.sql.execution.aggregate.TypedAggregateExpression * @tparam OUT The type of the final output result. * @since 1.6.0 */ +@SerialVersionUID(2093413866369130093L) Review Comment: It's for mapping the client's `Aggregator` class to this one. This is required because we now serialise the whole `Aggregator` instance on the client side. If we decide to serialise individual methods instead, this `SerialVersionUID` can be abolished. Wdyt? -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org - To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org For additional commands, e-mail: reviews-h...@spark.apache.org
Re: [PR] [SPARK-48008][WIP] Support UDAFs in Spark Connect [spark]
hvanhovell commented on code in PR #46245:
URL: https://github.com/apache/spark/pull/46245#discussion_r1594544696
##
connector/connect/client/jvm/src/main/scala/org/apache/spark/sql/expressions/Aggregator.scala:
##
@@ -0,0 +1,104 @@
+/*
+ * 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.spark.sql.expressions
+
+import org.apache.spark.sql.{Encoder, TypedColumn}
+
+/**
+ * A base class for user-defined aggregations, which can be used in `Dataset`
operations to take
+ * all of the elements of a group and reduce them to a single value.
+ *
+ * For example, the following aggregator extracts an `int` from a specific
class and adds them up:
+ * {{{
+ * case class Data(i: Int)
+ *
+ * val customSummer = new Aggregator[Data, Int, Int] {
+ * def zero: Int = 0
+ * def reduce(b: Int, a: Data): Int = b + a.i
+ * def merge(b1: Int, b2: Int): Int = b1 + b2
+ * def finish(r: Int): Int = r
+ * def bufferEncoder: Encoder[Int] = Encoders.scalaInt
+ * def outputEncoder: Encoder[Int] = Encoders.scalaInt
+ * }
+ *
+ * spark.udf.register("customSummer", udaf(customSummer))
+ * val ds: Dataset[Data] = ...
+ * val aggregated = ds.selectExpr("customSummer(i)")
+ * }}}
+ *
+ * Based loosely on Aggregator from Algebird:
https://github.com/twitter/algebird
+ *
+ * @tparam IN The input type for the aggregation.
+ * @tparam BUF The type of the intermediate value of the reduction.
+ * @tparam OUT The type of the final output result.
+ * @since 4.0.0
+ */
+@SerialVersionUID(2093413866369130093L)
+abstract class Aggregator[-IN, BUF, OUT] extends Serializable {
+
+ /**
+ * A zero value for this aggregation. Should satisfy the property that any b
+ zero = b.
+ *
+ * @since 4.0.0
+ */
+ def zero: BUF
+
+ /**
+ * Combine two values to produce a new value. For performance, the function
may modify `b` and
+ * return it instead of constructing new object for b.
+ *
+ * @since 4.0.0
+ */
+ def reduce(b: BUF, a: IN): BUF
+
+ /**
+ * Merge two intermediate values.
+ *
+ * @since 4.0.0
+ */
+ def merge(b1: BUF, b2: BUF): BUF
+
+ /**
+ * Transform the output of the reduction.
+ *
+ * @since 4.0.0
+ */
+ def finish(reduction: BUF): OUT
+
+ /**
+ * Specifies the `Encoder` for the intermediate value type.
+ *
+ * @since 4.0.0
+ */
+ def bufferEncoder: Encoder[BUF]
+
+ /**
+ * Specifies the `Encoder` for the final output value type.
+ *
+ * @since 4.0.0
+ */
+ def outputEncoder: Encoder[OUT]
+
+ /**
+ * Returns this `Aggregator` as a `TypedColumn` that can be used in
`Dataset`.
+ * operations.
+ */
+ def toColumn: TypedColumn[IN, OUT] = {
Review Comment:
I think it should take a wildcard, we have done similar things before.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
Re: [PR] [SPARK-48008][WIP] Support UDAFs in Spark Connect [spark]
hvanhovell commented on code in PR #46245:
URL: https://github.com/apache/spark/pull/46245#discussion_r1594250192
##
connector/connect/client/jvm/src/main/scala/org/apache/spark/sql/expressions/Aggregator.scala:
##
@@ -0,0 +1,104 @@
+/*
+ * 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.spark.sql.expressions
+
+import org.apache.spark.sql.{Encoder, TypedColumn}
+
+/**
+ * A base class for user-defined aggregations, which can be used in `Dataset`
operations to take
+ * all of the elements of a group and reduce them to a single value.
+ *
+ * For example, the following aggregator extracts an `int` from a specific
class and adds them up:
+ * {{{
+ * case class Data(i: Int)
+ *
+ * val customSummer = new Aggregator[Data, Int, Int] {
+ * def zero: Int = 0
+ * def reduce(b: Int, a: Data): Int = b + a.i
+ * def merge(b1: Int, b2: Int): Int = b1 + b2
+ * def finish(r: Int): Int = r
+ * def bufferEncoder: Encoder[Int] = Encoders.scalaInt
+ * def outputEncoder: Encoder[Int] = Encoders.scalaInt
+ * }
+ *
+ * spark.udf.register("customSummer", udaf(customSummer))
+ * val ds: Dataset[Data] = ...
+ * val aggregated = ds.selectExpr("customSummer(i)")
+ * }}}
+ *
+ * Based loosely on Aggregator from Algebird:
https://github.com/twitter/algebird
+ *
+ * @tparam IN The input type for the aggregation.
+ * @tparam BUF The type of the intermediate value of the reduction.
+ * @tparam OUT The type of the final output result.
+ * @since 4.0.0
+ */
+@SerialVersionUID(2093413866369130093L)
+abstract class Aggregator[-IN, BUF, OUT] extends Serializable {
+
+ /**
+ * A zero value for this aggregation. Should satisfy the property that any b
+ zero = b.
+ *
+ * @since 4.0.0
+ */
+ def zero: BUF
+
+ /**
+ * Combine two values to produce a new value. For performance, the function
may modify `b` and
+ * return it instead of constructing new object for b.
+ *
+ * @since 4.0.0
+ */
+ def reduce(b: BUF, a: IN): BUF
+
+ /**
+ * Merge two intermediate values.
+ *
+ * @since 4.0.0
+ */
+ def merge(b1: BUF, b2: BUF): BUF
+
+ /**
+ * Transform the output of the reduction.
+ *
+ * @since 4.0.0
+ */
+ def finish(reduction: BUF): OUT
+
+ /**
+ * Specifies the `Encoder` for the intermediate value type.
+ *
+ * @since 4.0.0
+ */
+ def bufferEncoder: Encoder[BUF]
+
+ /**
+ * Specifies the `Encoder` for the final output value type.
+ *
+ * @since 4.0.0
+ */
+ def outputEncoder: Encoder[OUT]
+
+ /**
+ * Returns this `Aggregator` as a `TypedColumn` that can be used in
`Dataset`.
+ * operations.
+ */
+ def toColumn: TypedColumn[IN, OUT] = {
Review Comment:
How should this work on the connect side?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
Re: [PR] [SPARK-48008][WIP] Support UDAFs in Spark Connect [spark]
hvanhovell commented on code in PR #46245:
URL: https://github.com/apache/spark/pull/46245#discussion_r1594229607
##
connector/connect/common/src/main/protobuf/spark/connect/expressions.proto:
##
@@ -379,6 +380,15 @@ message ScalarScalaUDF {
bool nullable = 4;
}
+message ScalaUDAF {
+ // (Required) Serialized JVM object of the Aggregator instance, including
buffer and output encoders
+ bytes payload = 1;
+ // (Required) Input type of the UDAF
+ DataType inputType = 2;
+ // (Required) True if the UDAF can return null value
+ bool nullable = 3;
Review Comment:
For the record we should not rely on the payload for this.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
Re: [PR] [SPARK-48008][WIP] Support UDAFs in Spark Connect [spark]
hvanhovell commented on code in PR #46245:
URL: https://github.com/apache/spark/pull/46245#discussion_r1594227631
##
connector/connect/common/src/main/protobuf/spark/connect/expressions.proto:
##
@@ -379,6 +380,15 @@ message ScalarScalaUDF {
bool nullable = 4;
}
+message ScalaUDAF {
+ // (Required) Serialized JVM object of the Aggregator instance, including
buffer and output encoders
+ bytes payload = 1;
+ // (Required) Input type of the UDAF
+ DataType inputType = 2;
+ // (Required) True if the UDAF can return null value
+ bool nullable = 3;
Review Comment:
Why is the return type not needed?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
Re: [PR] [SPARK-48008][WIP] Support UDAFs in Spark Connect [spark]
hvanhovell commented on code in PR #46245: URL: https://github.com/apache/spark/pull/46245#discussion_r1594223201 ## sql/core/src/main/scala/org/apache/spark/sql/expressions/Aggregator.scala: ## @@ -49,6 +49,7 @@ import org.apache.spark.sql.execution.aggregate.TypedAggregateExpression * @tparam OUT The type of the final output result. * @since 1.6.0 */ +@SerialVersionUID(2093413866369130093L) Review Comment: `TypedColumn`? -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org - To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org For additional commands, e-mail: reviews-h...@spark.apache.org
Re: [PR] [SPARK-48008][WIP] Support UDAFs in Spark Connect [spark]
xupefei commented on code in PR #46245:
URL: https://github.com/apache/spark/pull/46245#discussion_r1594216679
##
connector/connect/client/jvm/src/main/scala/org/apache/spark/sql/expressions/Aggregator.scala:
##
@@ -0,0 +1,104 @@
+/*
+ * 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.spark.sql.expressions
+
+import org.apache.spark.sql.{Encoder, TypedColumn}
+
+/**
+ * A base class for user-defined aggregations, which can be used in `Dataset`
operations to take
+ * all of the elements of a group and reduce them to a single value.
+ *
+ * For example, the following aggregator extracts an `int` from a specific
class and adds them up:
+ * {{{
+ * case class Data(i: Int)
+ *
+ * val customSummer = new Aggregator[Data, Int, Int] {
+ * def zero: Int = 0
+ * def reduce(b: Int, a: Data): Int = b + a.i
+ * def merge(b1: Int, b2: Int): Int = b1 + b2
+ * def finish(r: Int): Int = r
+ * def bufferEncoder: Encoder[Int] = Encoders.scalaInt
+ * def outputEncoder: Encoder[Int] = Encoders.scalaInt
+ * }
+ *
+ * spark.udf.register("customSummer", udaf(customSummer))
+ * val ds: Dataset[Data] = ...
+ * val aggregated = ds.selectExpr("customSummer(i)")
+ * }}}
+ *
+ * Based loosely on Aggregator from Algebird:
https://github.com/twitter/algebird
+ *
+ * @tparam IN The input type for the aggregation.
+ * @tparam BUF The type of the intermediate value of the reduction.
+ * @tparam OUT The type of the final output result.
+ * @since 4.0.0
+ */
+@SerialVersionUID(2093413866369130093L)
+abstract class Aggregator[-IN, BUF, OUT] extends Serializable {
Review Comment:
Yes, that would be ideal. I was doing that before until I found that the
Connect client should have another docstring `@since 4.0.0`. Could you suggest
how could we document this on the client side if this class is moved to Common?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
Re: [PR] [SPARK-48008][WIP] Support UDAFs in Spark Connect [spark]
hvanhovell commented on code in PR #46245:
URL: https://github.com/apache/spark/pull/46245#discussion_r1594210664
##
connector/connect/client/jvm/src/main/scala/org/apache/spark/sql/expressions/UserDefinedFunction.scala:
##
@@ -205,6 +205,91 @@ object ScalarUserDefinedFunction {
}
}
+case class UserDefinedAggregationFunction private[sql] (
Review Comment:
How much of this implementation is shared with the
`ScalarUserDefinedFunction`?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
Re: [PR] [SPARK-48008][WIP] Support UDAFs in Spark Connect [spark]
hvanhovell commented on code in PR #46245:
URL: https://github.com/apache/spark/pull/46245#discussion_r1594209214
##
connector/connect/client/jvm/src/main/scala/org/apache/spark/sql/expressions/Aggregator.scala:
##
@@ -0,0 +1,104 @@
+/*
+ * 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.spark.sql.expressions
+
+import org.apache.spark.sql.{Encoder, TypedColumn}
+
+/**
+ * A base class for user-defined aggregations, which can be used in `Dataset`
operations to take
+ * all of the elements of a group and reduce them to a single value.
+ *
+ * For example, the following aggregator extracts an `int` from a specific
class and adds them up:
+ * {{{
+ * case class Data(i: Int)
+ *
+ * val customSummer = new Aggregator[Data, Int, Int] {
+ * def zero: Int = 0
+ * def reduce(b: Int, a: Data): Int = b + a.i
+ * def merge(b1: Int, b2: Int): Int = b1 + b2
+ * def finish(r: Int): Int = r
+ * def bufferEncoder: Encoder[Int] = Encoders.scalaInt
+ * def outputEncoder: Encoder[Int] = Encoders.scalaInt
+ * }
+ *
+ * spark.udf.register("customSummer", udaf(customSummer))
+ * val ds: Dataset[Data] = ...
+ * val aggregated = ds.selectExpr("customSummer(i)")
+ * }}}
+ *
+ * Based loosely on Aggregator from Algebird:
https://github.com/twitter/algebird
+ *
+ * @tparam IN The input type for the aggregation.
+ * @tparam BUF The type of the intermediate value of the reduction.
+ * @tparam OUT The type of the final output result.
+ * @since 4.0.0
+ */
+@SerialVersionUID(2093413866369130093L)
+abstract class Aggregator[-IN, BUF, OUT] extends Serializable {
Review Comment:
Can we move this common instead of having two abstract classes?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
Re: [PR] [SPARK-48008][WIP] Support UDAFs in Spark Connect [spark]
hvanhovell commented on code in PR #46245: URL: https://github.com/apache/spark/pull/46245#discussion_r1594208101 ## sql/core/src/main/scala/org/apache/spark/sql/expressions/Aggregator.scala: ## @@ -49,6 +49,7 @@ import org.apache.spark.sql.execution.aggregate.TypedAggregateExpression * @tparam OUT The type of the final output result. * @since 1.6.0 */ +@SerialVersionUID(2093413866369130093L) Review Comment: Why is this needed? -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org - To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org For additional commands, e-mail: reviews-h...@spark.apache.org
