[jira] [Assigned] (SPARK-11594) Cannot create UDAF in REPL
[
https://issues.apache.org/jira/browse/SPARK-11594?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Apache Spark reassigned SPARK-11594:
Assignee: (was: Apache Spark)
> Cannot create UDAF in REPL
> --
>
> Key: SPARK-11594
> URL: https://issues.apache.org/jira/browse/SPARK-11594
> Project: Spark
> Issue Type: Bug
> Components: SQL
>Affects Versions: 1.5.1, 1.6.0
> Environment: Latest Spark Master
> JVM 1.8.0_66-b17
>Reporter: Herman van Hovell
>Priority: Minor
>
> If you try to define the a UDAF in the REPL, an internal error is thrown by
> Java. The following code for example:
> {noformat}
> import org.apache.spark.sql.Row
> import org.apache.spark.sql.types.{DataType, LongType, StructType}
> import org.apache.spark.sql.expressions.{MutableAggregationBuffer,
> UserDefinedAggregateFunction}
> class LongProductSum extends UserDefinedAggregateFunction {
> def inputSchema: StructType = new StructType()
> .add("a", LongType)
> .add("b", LongType)
> def bufferSchema: StructType = new StructType()
> .add("product", LongType)
> def dataType: DataType = LongType
> def deterministic: Boolean = true
> def initialize(buffer: MutableAggregationBuffer): Unit = {
> buffer(0) = 0L
> }
> def update(buffer: MutableAggregationBuffer, input: Row): Unit = {
> if (!(input.isNullAt(0) || input.isNullAt(1))) {
> buffer(0) = buffer.getLong(0) + input.getLong(0) * input.getLong(1)
> }
> }
> def merge(buffer1: MutableAggregationBuffer, buffer2: Row): Unit = {
> buffer1(0) = buffer1.getLong(0) + buffer2.getLong(0)
> }
> def evaluate(buffer: Row): Any =
> buffer.getLong(0)
> }
> sqlContext.udf.register("longProductSum", new LongProductSum)
> val data2 = Seq[(Integer, Integer, Integer)](
> (1, 10, -10),
> (null, -60, 60),
> (1, 30, -30),
> (1, 30, 30),
> (2, 1, 1),
> (3, null, null)).toDF("key", "value1", "value2")
> data2.registerTempTable("agg2")
> val q = sqlContext.sql("""
> |SELECT
> | key,
> | count(distinct value1, value2),
> | longProductSum(distinct value1, value2)
> |FROM agg2
> |GROUP BY key
> """.stripMargin)
> q.show
> {noformat}
> Will throw the following error:
> {noformat}
> java.lang.InternalError: Malformed class name
> at java.lang.Class.getSimpleName(Class.java:1330)
> at
> org.apache.spark.sql.execution.aggregate.ScalaUDAF.toString(udaf.scala:455)
> at
> org.apache.spark.sql.execution.SparkStrategies$Aggregation$$anonfun$9.apply(SparkStrategies.scala:211)
> at
> org.apache.spark.sql.execution.SparkStrategies$Aggregation$$anonfun$9.apply(SparkStrategies.scala:209)
> at
> scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:244)
> at
> scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:244)
> at
> scala.collection.mutable.ResizableArray$class.foreach(ResizableArray.scala:59)
> at scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:47)
> at scala.collection.TraversableLike$class.map(TraversableLike.scala:244)
> at scala.collection.AbstractTraversable.map(Traversable.scala:105)
> at
> org.apache.spark.sql.execution.SparkStrategies$Aggregation$.apply(SparkStrategies.scala:209)
> at
> org.apache.spark.sql.catalyst.planning.QueryPlanner$$anonfun$1.apply(QueryPlanner.scala:58)
> at
> org.apache.spark.sql.catalyst.planning.QueryPlanner$$anonfun$1.apply(QueryPlanner.scala:58)
> at scala.collection.Iterator$$anon$13.hasNext(Iterator.scala:371)
> at
> org.apache.spark.sql.catalyst.planning.QueryPlanner.plan(QueryPlanner.scala:59)
> at
> org.apache.spark.sql.catalyst.planning.QueryPlanner.planLater(QueryPlanner.scala:54)
> at
> org.apache.spark.sql.execution.SparkStrategies$BasicOperators$.apply(SparkStrategies.scala:445)
> at
> org.apache.spark.sql.catalyst.planning.QueryPlanner$$anonfun$1.apply(QueryPlanner.scala:58)
> at
> org.apache.spark.sql.catalyst.planning.QueryPlanner$$anonfun$1.apply(QueryPlanner.scala:58)
> at scala.collection.Iterator$$anon$13.hasNext(Iterator.scala:371)
> at
> org.apache.spark.sql.catalyst.planning.QueryPlanner.plan(QueryPlanner.scala:59)
> at
> org.apache.spark.sql.execution.QueryExecution.sparkPlan$lzycompute(QueryExecution.scala:51)
> at
> org.apache.spark.sql.execution.QueryExecution.sparkPlan(QueryExecution.scala:49)
> at
> org.apache.spark.sql.execution.QueryExecution.executedPlan$lzycompute(QueryExecution.scala:56)
> at
> org.apache.spark.sql.execution.QueryExecution.executedPlan(QueryExecution.scala:56)
> at org.apache.spark.sql.DataFrame.withCallback(DataFrame.scala:2092)
> at org.apache.spark.sql.DataFrame.head(DataFrame.scala:1419)
> at org.apache.spark
[jira] [Assigned] (SPARK-11594) Cannot create UDAF in REPL
[
https://issues.apache.org/jira/browse/SPARK-11594?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Apache Spark reassigned SPARK-11594:
Assignee: Apache Spark
> Cannot create UDAF in REPL
> --
>
> Key: SPARK-11594
> URL: https://issues.apache.org/jira/browse/SPARK-11594
> Project: Spark
> Issue Type: Bug
> Components: SQL
>Affects Versions: 1.5.1, 1.6.0
> Environment: Latest Spark Master
> JVM 1.8.0_66-b17
>Reporter: Herman van Hovell
>Assignee: Apache Spark
>Priority: Minor
>
> If you try to define the a UDAF in the REPL, an internal error is thrown by
> Java. The following code for example:
> {noformat}
> import org.apache.spark.sql.Row
> import org.apache.spark.sql.types.{DataType, LongType, StructType}
> import org.apache.spark.sql.expressions.{MutableAggregationBuffer,
> UserDefinedAggregateFunction}
> class LongProductSum extends UserDefinedAggregateFunction {
> def inputSchema: StructType = new StructType()
> .add("a", LongType)
> .add("b", LongType)
> def bufferSchema: StructType = new StructType()
> .add("product", LongType)
> def dataType: DataType = LongType
> def deterministic: Boolean = true
> def initialize(buffer: MutableAggregationBuffer): Unit = {
> buffer(0) = 0L
> }
> def update(buffer: MutableAggregationBuffer, input: Row): Unit = {
> if (!(input.isNullAt(0) || input.isNullAt(1))) {
> buffer(0) = buffer.getLong(0) + input.getLong(0) * input.getLong(1)
> }
> }
> def merge(buffer1: MutableAggregationBuffer, buffer2: Row): Unit = {
> buffer1(0) = buffer1.getLong(0) + buffer2.getLong(0)
> }
> def evaluate(buffer: Row): Any =
> buffer.getLong(0)
> }
> sqlContext.udf.register("longProductSum", new LongProductSum)
> val data2 = Seq[(Integer, Integer, Integer)](
> (1, 10, -10),
> (null, -60, 60),
> (1, 30, -30),
> (1, 30, 30),
> (2, 1, 1),
> (3, null, null)).toDF("key", "value1", "value2")
> data2.registerTempTable("agg2")
> val q = sqlContext.sql("""
> |SELECT
> | key,
> | count(distinct value1, value2),
> | longProductSum(distinct value1, value2)
> |FROM agg2
> |GROUP BY key
> """.stripMargin)
> q.show
> {noformat}
> Will throw the following error:
> {noformat}
> java.lang.InternalError: Malformed class name
> at java.lang.Class.getSimpleName(Class.java:1330)
> at
> org.apache.spark.sql.execution.aggregate.ScalaUDAF.toString(udaf.scala:455)
> at
> org.apache.spark.sql.execution.SparkStrategies$Aggregation$$anonfun$9.apply(SparkStrategies.scala:211)
> at
> org.apache.spark.sql.execution.SparkStrategies$Aggregation$$anonfun$9.apply(SparkStrategies.scala:209)
> at
> scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:244)
> at
> scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:244)
> at
> scala.collection.mutable.ResizableArray$class.foreach(ResizableArray.scala:59)
> at scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:47)
> at scala.collection.TraversableLike$class.map(TraversableLike.scala:244)
> at scala.collection.AbstractTraversable.map(Traversable.scala:105)
> at
> org.apache.spark.sql.execution.SparkStrategies$Aggregation$.apply(SparkStrategies.scala:209)
> at
> org.apache.spark.sql.catalyst.planning.QueryPlanner$$anonfun$1.apply(QueryPlanner.scala:58)
> at
> org.apache.spark.sql.catalyst.planning.QueryPlanner$$anonfun$1.apply(QueryPlanner.scala:58)
> at scala.collection.Iterator$$anon$13.hasNext(Iterator.scala:371)
> at
> org.apache.spark.sql.catalyst.planning.QueryPlanner.plan(QueryPlanner.scala:59)
> at
> org.apache.spark.sql.catalyst.planning.QueryPlanner.planLater(QueryPlanner.scala:54)
> at
> org.apache.spark.sql.execution.SparkStrategies$BasicOperators$.apply(SparkStrategies.scala:445)
> at
> org.apache.spark.sql.catalyst.planning.QueryPlanner$$anonfun$1.apply(QueryPlanner.scala:58)
> at
> org.apache.spark.sql.catalyst.planning.QueryPlanner$$anonfun$1.apply(QueryPlanner.scala:58)
> at scala.collection.Iterator$$anon$13.hasNext(Iterator.scala:371)
> at
> org.apache.spark.sql.catalyst.planning.QueryPlanner.plan(QueryPlanner.scala:59)
> at
> org.apache.spark.sql.execution.QueryExecution.sparkPlan$lzycompute(QueryExecution.scala:51)
> at
> org.apache.spark.sql.execution.QueryExecution.sparkPlan(QueryExecution.scala:49)
> at
> org.apache.spark.sql.execution.QueryExecution.executedPlan$lzycompute(QueryExecution.scala:56)
> at
> org.apache.spark.sql.execution.QueryExecution.executedPlan(QueryExecution.scala:56)
> at org.apache.spark.sql.DataFrame.withCallback(DataFrame.scala:2092)
> at org.apache.spark.sql.DataFrame.head(DataFrame.scala:1419)
>
