Mailing-List: contact issues-help@spark.apache.org; run by ezmlm
Precedence: bulk
Date: Mon, 9 Nov 2015 13:24:11 +0000 (UTC)
From: "Apache Spark (JIRA)" <jira@apache.org>
To: issues@spark.apache.org
Message-ID: <JIRA.12911511.1447069216000.15422.1447075451116@Atlassian.JIRA>
In-Reply-To: <JIRA.12911511.1447069216000@Atlassian.JIRA>
References: <JIRA.12911511.1447069216000@Atlassian.JIRA>
 <JIRA.12911511.1447069216076@arcas>
Subject: [jira] [Assigned] (SPARK-11594) Cannot create UDAF in REPL
MIME-Version: 1.0
Content-Type: text/plain; charset=utf-8
Content-Transfer-Encoding: 7bit


     [ 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)
> 	at org.apache.spark.sql.DataFrame.take(DataFrame.scala:1488)
> 	at org.apache.spark.sql.DataFrame.showString(DataFrame.scala:171)
> 	at org.apache.spark.sql.DataFrame.show(DataFrame.scala:404)
> 	at org.apache.spark.sql.DataFrame.show(DataFrame.scala:365)
> 	at org.apache.spark.sql.DataFrame.show(DataFrame.scala:373)
> 	at .<init>(<console>:52)
> 	at .<clinit>(<console>)
> 	at .<init>(<console>:7)
> 	at .<clinit>(<console>)
> 	at $print(<console>)
> 	at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
> 	at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
> 	at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
> 	at java.lang.reflect.Method.invoke(Method.java:497)
> 	at scala.tools.nsc.interpreter.IMain$ReadEvalPrint.call(IMain.scala:734)
> 	at scala.tools.nsc.interpreter.IMain$Request.loadAndRun(IMain.scala:983)
> 	at scala.tools.nsc.interpreter.IMain.loadAndRunReq$1(IMain.scala:573)
> 	at scala.tools.nsc.interpreter.IMain.interpret(IMain.scala:604)
> 	at scala.tools.nsc.interpreter.IMain.interpret(IMain.scala:568)
> 	at scala.tools.nsc.interpreter.ILoop.reallyInterpret$1(ILoop.scala:760)
> 	at scala.tools.nsc.interpreter.ILoop.interpretStartingWith(ILoop.scala:805)
> 	at scala.tools.nsc.interpreter.ILoop.command(ILoop.scala:717)
> 	at scala.tools.nsc.interpreter.ILoop.processLine$1(ILoop.scala:581)
> 	at scala.tools.nsc.interpreter.ILoop.innerLoop$1(ILoop.scala:588)
> 	at scala.tools.nsc.interpreter.ILoop.loop(ILoop.scala:591)
> 	at scala.tools.nsc.interpreter.ILoop$$anonfun$process$1.apply$mcZ$sp(ILoop.scala:882)
> 	at scala.tools.nsc.interpreter.ILoop$$anonfun$process$1.apply(ILoop.scala:837)
> 	at scala.tools.nsc.interpreter.ILoop$$anonfun$process$1.apply(ILoop.scala:837)
> 	at scala.tools.nsc.util.ScalaClassLoader$.savingContextLoader(ScalaClassLoader.scala:135)
> 	at scala.tools.nsc.interpreter.ILoop.process(ILoop.scala:837)
> 	at scala.tools.nsc.interpreter.ILoop.main(ILoop.scala:904)
> 	at xsbt.ConsoleInterface.run(ConsoleInterface.scala:62)
> 	at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
> 	at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
> 	at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
> 	at java.lang.reflect.Method.invoke(Method.java:497)
> 	at sbt.compiler.AnalyzingCompiler.call(AnalyzingCompiler.scala:101)
> 	at sbt.compiler.AnalyzingCompiler.console(AnalyzingCompiler.scala:76)
> 	at sbt.Console.sbt$Console$$console0$1(Console.scala:22)
> 	at sbt.Console$$anonfun$apply$2$$anonfun$apply$1.apply$mcV$sp(Console.scala:23)
> 	at sbt.Console$$anonfun$apply$2$$anonfun$apply$1.apply(Console.scala:23)
> 	at sbt.Console$$anonfun$apply$2$$anonfun$apply$1.apply(Console.scala:23)
> 	at sbt.Logger$$anon$4.apply(Logger.scala:85)
> 	at sbt.TrapExit$App.run(TrapExit.scala:248)
> 	at java.lang.Thread.run(Thread.java:745)
> {noformat}
> Seems like my JVM (1.8.0_66-b17) does not like REPL created classes.


--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

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