You are viewing a plain text version of this content. The canonical link for it is here.
Posted to user@spark.apache.org by Lee Becker <le...@hapara.com> on 2016/04/23 00:45:40 UTC
Dataset aggregateByKey equivalent
Is there a way to do aggregateByKey on Datasets the way one can on an RDD?
Consider the following RDD code to build a set of KeyVals into a DataFrame
containing a column with the KeyVals' keys and a column containing lists of
KeyVals. The end goal is to join it with collections which which will be
similarly transformed.
case class KeyVal(k: Int, v: Int)
val keyVals = sc.parallelize(for (i <- 1 to 3; j <- 4 to 6) yield KeyVal(i,j))
// function for appending to list
val addToList = (s: List[KeyVal], v: KeyVal) => s :+ v
// function for merging two lists
val addLists = (s: List[KeyVal], t: List[KeyVal]) => s++t
val keyAndKeyVals = keyVals.map(kv=> (kv.k, kv))
val keyAndNestedKeyVals = keyAndKeyVals.
aggregateByKey(List[KeyVal]())(addToList, addLists).
toDF("key", "keyvals")
keyAndNestedKeyVals.show
which produces:
+---+--------------------+
|key| keyvals|
+---+--------------------+
| 1|[[1,4], [1,5], [1...|
| 2|[[2,4], [2,5], [2...|
| 3|[[3,4], [3,5], [3...|
+---+--------------------+
For a Dataset approach I tried the following to no avail:
// Initialize as Dataset
val keyVals = sc.parallelize(for (i <- 1 to 3; j <- 4 to 6) yield KeyVal(i,j)).
toDS
// Build key, keyVal mappings
val keyValsByKey = keyVals.groupBy(kv=>(kv.k))
case class NestedKeyVal(key: Int, keyvals: List[KeyVal])
val convertToNested = (key: Int, keyValsIter: Iterator[KeyVal]) =>
NestedKeyVal(key=key, keyvals=keyValsIter.toList)
val keyValsNestedByKey = keyValsByKey.mapGroups((key,keyvals) =>
convertToNested(key,keyvals))
keyValsNestedByKey.show
This and several other incantations using groupBy + mapGroups consistently
gives me serialization problems. Is this because the iterator can not be
guaranteed across boundaries?
Or is there some issue with what a Dataset can encode in the interim. What
other ways might I approach this problem?
Thanks,
Lee
Re: Dataset aggregateByKey equivalent
Posted by Lee Becker <le...@hapara.com>.
On Sat, Apr 23, 2016 at 8:56 AM, Michael Armbrust <mi...@databricks.com>
wrote:
> Have you looked at aggregators?
>
>
> https://docs.cloud.databricks.com/docs/spark/1.6/index.html#examples/Dataset%20Aggregator.html
>
Thanks for the pointer to aggregators. I wasn't yet aware of them.
However, I still get similar errors when attempting to go the Dataset
route. Here is my attempt at an aggregator class for the example above.
import org.apache.spark.sql.expressions.Aggregator
case class KeyVal(k: Int, v: Int)
val keyValsDs = sc.parallelize(for (i <- 1 to 3; j <- 4 to 6) yield
KeyVal(i,j)).toDS
class AggKV extends Aggregator[KeyVal, List[KeyVal], List[KeyVal]]
with Serializable {
override def zero: List[KeyVal] = List()
override def reduce(b: List[KeyVal], a: KeyVal): List[KeyVal] = b :+ a
override def finish(reduction: List[KeyVal]): List[KeyVal] = reduction
override def merge(b1: List[KeyVal], b2: List[KeyVal]): List[KeyVal]
= b1 ++ b2
}
The following shows production of the correct schema
keyValsDs.groupBy($"k").agg(new AggKV().toColumn)
org.apache.spark.sql.Dataset[(org.apache.spark.sql.Row, List[KeyVal])] =
[_1: struct<k:int>, _2: struct<value:array<struct<k:int,v:int>>>]
Actual execution fails with Task not serializable. Am I missing something
or is this just not possible without dropping into RDDs?
scala> keyValsDs.groupBy($"k").agg(new AggKV().toColumn).show
org.apache.spark.sql.catalyst.errors.package$TreeNodeException: execute,
tree:
SortBasedAggregate(key=[k#684],
functions=[(AggKV(k#684,v#685),mode=Final,isDistinct=false)],
output=[key#739,AggKV(k,v)#738])
+- ConvertToSafe
+- Sort [k#684 ASC], false, 0
+- TungstenExchange hashpartitioning(k#684,200), None
+- ConvertToUnsafe
+- SortBasedAggregate(key=[k#684],
functions=[(AggKV(k#684,v#685),mode=Partial,isDistinct=false)],
output=[k#684,value#731])
+- ConvertToSafe
+- Sort [k#684 ASC], false, 0
+- Project [k#684,v#685,k#684]
+- Scan ExistingRDD[k#684,v#685]
at
org.apache.spark.sql.catalyst.errors.package$.attachTree(package.scala:49)
at
org.apache.spark.sql.execution.aggregate.SortBasedAggregate.doExecute(SortBasedAggregate.scala:69)
at
org.apache.spark.sql.execution.SparkPlan$$anonfun$execute$5.apply(SparkPlan.scala:132)
at
org.apache.spark.sql.execution.SparkPlan$$anonfun$execute$5.apply(SparkPlan.scala:130)
at
org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:150)
at
org.apache.spark.sql.execution.SparkPlan.execute(SparkPlan.scala:130)
at
org.apache.spark.sql.execution.SparkPlan.executeTake(SparkPlan.scala:187)
at
org.apache.spark.sql.execution.Limit.executeCollect(basicOperators.scala:165)
at
org.apache.spark.sql.execution.SparkPlan.executeCollectPublic(SparkPlan.scala:174)
at
org.apache.spark.sql.DataFrame$$anonfun$org$apache$spark$sql$DataFrame$$execute$1$1.apply(DataFrame.scala:1538)
at
org.apache.spark.sql.DataFrame$$anonfun$org$apache$spark$sql$DataFrame$$execute$1$1.apply(DataFrame.scala:1538)
at
org.apache.spark.sql.execution.SQLExecution$.withNewExecutionId(SQLExecution.scala:56)
at
org.apache.spark.sql.DataFrame.withNewExecutionId(DataFrame.scala:2125)
at org.apache.spark.sql.DataFrame.org
$apache$spark$sql$DataFrame$$execute$1(DataFrame.scala:1537)
at org.apache.spark.sql.DataFrame.org
$apache$spark$sql$DataFrame$$collect(DataFrame.scala:1544)
at
org.apache.spark.sql.DataFrame$$anonfun$head$1.apply(DataFrame.scala:1414)
at
org.apache.spark.sql.DataFrame$$anonfun$head$1.apply(DataFrame.scala:1413)
at org.apache.spark.sql.DataFrame.withCallback(DataFrame.scala:2138)
at org.apache.spark.sql.DataFrame.head(DataFrame.scala:1413)
at org.apache.spark.sql.DataFrame.take(DataFrame.scala:1495)
at org.apache.spark.sql.DataFrame.showString(DataFrame.scala:171)
at org.apache.spark.sql.DataFrame.show(DataFrame.scala:394)
at org.apache.spark.sql.Dataset.show(Dataset.scala:228)
at org.apache.spark.sql.Dataset.show(Dataset.scala:192)
at org.apache.spark.sql.Dataset.show(Dataset.scala:200)
at
$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC.<init>(<console>:50)
at
$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC.<init>(<console>:55)
at
$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC.<init>(<console>:57)
at
$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC.<init>(<console>:59)
at $iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC.<init>(<console>:61)
at $iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC.<init>(<console>:63)
at $iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC.<init>(<console>:65)
at $iwC$$iwC$$iwC$$iwC$$iwC$$iwC.<init>(<console>:67)
at $iwC$$iwC$$iwC$$iwC$$iwC.<init>(<console>:69)
at $iwC$$iwC$$iwC$$iwC.<init>(<console>:71)
at $iwC$$iwC$$iwC.<init>(<console>:73)
at $iwC$$iwC.<init>(<console>:75)
at $iwC.<init>(<console>:77)
at <init>(<console>:79)
at .<init>(<console>:83)
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
org.apache.spark.repl.SparkIMain$ReadEvalPrint.call(SparkIMain.scala:1065)
at
org.apache.spark.repl.SparkIMain$Request.loadAndRun(SparkIMain.scala:1346)
at
org.apache.spark.repl.SparkIMain.loadAndRunReq$1(SparkIMain.scala:840)
at org.apache.spark.repl.SparkIMain.interpret(SparkIMain.scala:871)
at org.apache.spark.repl.SparkIMain.interpret(SparkIMain.scala:819)
at
org.apache.spark.repl.SparkILoop.reallyInterpret$1(SparkILoop.scala:857)
at
org.apache.spark.repl.SparkILoop.interpretStartingWith(SparkILoop.scala:902)
at org.apache.spark.repl.SparkILoop.command(SparkILoop.scala:814)
at
org.apache.spark.repl.SparkILoop.processLine$1(SparkILoop.scala:657)
at
org.apache.spark.repl.SparkILoop.innerLoop$1(SparkILoop.scala:665)
at org.apache.spark.repl.SparkILoop.org
$apache$spark$repl$SparkILoop$$loop(SparkILoop.scala:670)
at
org.apache.spark.repl.SparkILoop$$anonfun$org$apache$spark$repl$SparkILoop$$process$1.apply$mcZ$sp(SparkILoop.scala:997)
at
org.apache.spark.repl.SparkILoop$$anonfun$org$apache$spark$repl$SparkILoop$$process$1.apply(SparkILoop.scala:945)
at
org.apache.spark.repl.SparkILoop$$anonfun$org$apache$spark$repl$SparkILoop$$process$1.apply(SparkILoop.scala:945)
at
scala.tools.nsc.util.ScalaClassLoader$.savingContextLoader(ScalaClassLoader.scala:135)
at org.apache.spark.repl.SparkILoop.org
$apache$spark$repl$SparkILoop$$process(SparkILoop.scala:945)
at org.apache.spark.repl.SparkILoop.process(SparkILoop.scala:1059)
at org.apache.spark.repl.Main$.main(Main.scala:31)
at org.apache.spark.repl.Main.main(Main.scala)
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_maven_executions.MainHelper.runMain(MainHelper.java:164)
at
scala_maven_executions.JavaMainCallerInProcess.runInternal(JavaMainCallerInProcess.java:94)
at
scala_maven_executions.JavaMainCallerInProcess.run(JavaMainCallerInProcess.java:54)
at
scala_maven_executions.JavaMainCallerSupport.run(JavaMainCallerSupport.java:110)
at scala_maven.ScalaConsoleMojo.doExecute(ScalaConsoleMojo.java:102)
at scala_maven.ScalaMojoSupport.execute(ScalaMojoSupport.java:482)
at
org.apache.maven.plugin.DefaultBuildPluginManager.executeMojo(DefaultBuildPluginManager.java:134)
at
org.apache.maven.lifecycle.internal.MojoExecutor.execute(MojoExecutor.java:208)
at
org.apache.maven.lifecycle.internal.MojoExecutor.execute(MojoExecutor.java:153)
at
org.apache.maven.lifecycle.internal.MojoExecutor.execute(MojoExecutor.java:145)
at
org.apache.maven.lifecycle.internal.LifecycleModuleBuilder.buildProject(LifecycleModuleBuilder.java:116)
at
org.apache.maven.lifecycle.internal.LifecycleModuleBuilder.buildProject(LifecycleModuleBuilder.java:80)
at
org.apache.maven.lifecycle.internal.builder.singlethreaded.SingleThreadedBuilder.build(SingleThreadedBuilder.java:51)
at
org.apache.maven.lifecycle.internal.LifecycleStarter.execute(LifecycleStarter.java:128)
at org.apache.maven.DefaultMaven.doExecute(DefaultMaven.java:307)
at org.apache.maven.DefaultMaven.doExecute(DefaultMaven.java:193)
at org.apache.maven.DefaultMaven.execute(DefaultMaven.java:106)
at org.apache.maven.cli.MavenCli.execute(MavenCli.java:862)
at org.apache.maven.cli.MavenCli.doMain(MavenCli.java:286)
at org.apache.maven.cli.MavenCli.main(MavenCli.java:197)
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
org.codehaus.plexus.classworlds.launcher.Launcher.launchEnhanced(Launcher.java:289)
at
org.codehaus.plexus.classworlds.launcher.Launcher.launch(Launcher.java:229)
at
org.codehaus.plexus.classworlds.launcher.Launcher.mainWithExitCode(Launcher.java:415)
at
org.codehaus.plexus.classworlds.launcher.Launcher.main(Launcher.java:356)
Caused by: org.apache.spark.sql.catalyst.errors.package$TreeNodeException:
execute, tree:
TungstenExchange hashpartitioning(k#684,200), None
+- ConvertToUnsafe
+- SortBasedAggregate(key=[k#684],
functions=[(AggKV(k#684,v#685),mode=Partial,isDistinct=false)],
output=[k#684,value#731])
+- ConvertToSafe
+- Sort [k#684 ASC], false, 0
+- Project [k#684,v#685,k#684]
+- Scan ExistingRDD[k#684,v#685]
at
org.apache.spark.sql.catalyst.errors.package$.attachTree(package.scala:49)
at
org.apache.spark.sql.execution.Exchange.doExecute(Exchange.scala:247)
at
org.apache.spark.sql.execution.SparkPlan$$anonfun$execute$5.apply(SparkPlan.scala:132)
at
org.apache.spark.sql.execution.SparkPlan$$anonfun$execute$5.apply(SparkPlan.scala:130)
at
org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:150)
at
org.apache.spark.sql.execution.SparkPlan.execute(SparkPlan.scala:130)
at org.apache.spark.sql.execution.Sort.doExecute(Sort.scala:64)
at
org.apache.spark.sql.execution.SparkPlan$$anonfun$execute$5.apply(SparkPlan.scala:132)
at
org.apache.spark.sql.execution.SparkPlan$$anonfun$execute$5.apply(SparkPlan.scala:130)
at
org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:150)
at
org.apache.spark.sql.execution.SparkPlan.execute(SparkPlan.scala:130)
at
org.apache.spark.sql.execution.ConvertToSafe.doExecute(rowFormatConverters.scala:56)
at
org.apache.spark.sql.execution.SparkPlan$$anonfun$execute$5.apply(SparkPlan.scala:132)
at
org.apache.spark.sql.execution.SparkPlan$$anonfun$execute$5.apply(SparkPlan.scala:130)
at
org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:150)
at
org.apache.spark.sql.execution.SparkPlan.execute(SparkPlan.scala:130)
at
org.apache.spark.sql.execution.aggregate.SortBasedAggregate$$anonfun$doExecute$1.apply(SortBasedAggregate.scala:72)
at
org.apache.spark.sql.execution.aggregate.SortBasedAggregate$$anonfun$doExecute$1.apply(SortBasedAggregate.scala:69)
at
org.apache.spark.sql.catalyst.errors.package$.attachTree(package.scala:48)
... 98 more
Caused by: org.apache.spark.SparkException: Task not serializable
at
org.apache.spark.util.ClosureCleaner$.ensureSerializable(ClosureCleaner.scala:304)
at
org.apache.spark.util.ClosureCleaner$.org$apache$spark$util$ClosureCleaner$$clean(ClosureCleaner.scala:294)
at
org.apache.spark.util.ClosureCleaner$.clean(ClosureCleaner.scala:122)
at org.apache.spark.SparkContext.clean(SparkContext.scala:2055)
at
org.apache.spark.rdd.RDD$$anonfun$mapPartitions$1.apply(RDD.scala:707)
at
org.apache.spark.rdd.RDD$$anonfun$mapPartitions$1.apply(RDD.scala:706)
at
org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:150)
at
org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:111)
at org.apache.spark.rdd.RDD.withScope(RDD.scala:316)
at org.apache.spark.rdd.RDD.mapPartitions(RDD.scala:706)
at
org.apache.spark.sql.execution.ConvertToUnsafe.doExecute(rowFormatConverters.scala:38)
at
org.apache.spark.sql.execution.SparkPlan$$anonfun$execute$5.apply(SparkPlan.scala:132)
at
org.apache.spark.sql.execution.SparkPlan$$anonfun$execute$5.apply(SparkPlan.scala:130)
at
org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:150)
at
org.apache.spark.sql.execution.SparkPlan.execute(SparkPlan.scala:130)
at
org.apache.spark.sql.execution.Exchange.prepareShuffleDependency(Exchange.scala:164)
at
org.apache.spark.sql.execution.Exchange$$anonfun$doExecute$1.apply(Exchange.scala:254)
at
org.apache.spark.sql.execution.Exchange$$anonfun$doExecute$1.apply(Exchange.scala:248)
at
org.apache.spark.sql.catalyst.errors.package$.attachTree(package.scala:48)
... 116 more
Caused by: java.io.NotSerializableException:
scala.reflect.internal.Mirrors$Roots$RootClass$
Serialization stack:
- object not serializable (class:
scala.reflect.internal.Mirrors$Roots$RootClass$, value: package <root>)
- field (class: scala.reflect.internal.Types$ThisType, name: sym,
type: class scala.reflect.internal.Symbols$Symbol)
- object (class scala.reflect.internal.Types$UniqueThisType, <root>)
- field (class: scala.reflect.internal.Types$SingleType, name: pre,
type: class scala.reflect.internal.Types$Type)
- object (class scala.reflect.internal.Types$UniqueSingleType,
$line142.type)
- field (class: scala.reflect.internal.Types$SingleType, name: pre,
type: class scala.reflect.internal.Types$Type)
- object (class scala.reflect.internal.Types$UniqueSingleType, type)
- field (class: scala.reflect.internal.Types$SingleType, name: pre,
type: class scala.reflect.internal.Types$Type)
- object (class scala.reflect.internal.Types$UniqueSingleType,
INSTANCE.type)
- field (class: scala.reflect.internal.Types$SingleType, name: pre,
type: class scala.reflect.internal.Types$Type)
- object (class scala.reflect.internal.Types$UniqueSingleType,
$iwC.type)
- field (class: scala.reflect.internal.Types$SingleType, name: pre,
type: class scala.reflect.internal.Types$Type)
- object (class scala.reflect.internal.Types$UniqueSingleType,
$iwC.$iwC.type)
- field (class: scala.reflect.internal.Types$TypeRef, name: pre,
type: class scala.reflect.internal.Types$Type)
- object (class scala.reflect.internal.Types$TypeRef$$anon$6,
KeyVal)
- field (class:
org.apache.spark.sql.catalyst.ScalaReflection$$anonfun$5, name:
elementType$2, type: class scala.reflect.api.Types$TypeApi)
- object (class
org.apache.spark.sql.catalyst.ScalaReflection$$anonfun$5, <function1>)
- field (class:
org.apache.spark.sql.catalyst.expressions.MapObjects, name: function, type:
interface scala.Function1)
- object (class
org.apache.spark.sql.catalyst.expressions.MapObjects,
mapobjects(<function1>,input[0,
ArrayType(StructType(StructField(k,IntegerType,false),
StructField(v,IntegerType,false)),true)],StructField(k,IntegerType,false),StructField(v,IntegerType,false)))
- field (class: org.apache.spark.sql.catalyst.expressions.Invoke,
name: targetObject, type: class
org.apache.spark.sql.catalyst.expressions.Expression)
- object (class org.apache.spark.sql.catalyst.expressions.Invoke,
invoke(mapobjects(<function1>,input[0,
ArrayType(StructType(StructField(k,IntegerType,false),
StructField(v,IntegerType,false)),true)],StructField(k,IntegerType,false),StructField(v,IntegerType,false)),array,ObjectType(class
[Ljava.lang.Object;)))
- writeObject data (class: scala.collection.immutable.$colon$colon)
- object (class scala.collection.immutable.$colon$colon,
List(invoke(mapobjects(<function1>,input[0,
ArrayType(StructType(StructField(k,IntegerType,false),
StructField(v,IntegerType,false)),true)],StructField(k,IntegerType,false),StructField(v,IntegerType,false)),array,ObjectType(class
[Ljava.lang.Object;))))
- field (class:
org.apache.spark.sql.catalyst.expressions.StaticInvoke, name: arguments,
type: interface scala.collection.Seq)
- object (class
org.apache.spark.sql.catalyst.expressions.StaticInvoke, staticinvoke(class
scala.collection.mutable.WrappedArray$,ObjectType(interface
scala.collection.Seq),make,invoke(mapobjects(<function1>,input[0,
ArrayType(StructType(StructField(k,IntegerType,false),
StructField(v,IntegerType,false)),true)],StructField(k,IntegerType,false),StructField(v,IntegerType,false)),array,ObjectType(class
[Ljava.lang.Object;)),true))
- field (class:
org.apache.spark.sql.catalyst.encoders.ExpressionEncoder, name:
fromRowExpression, type: class
org.apache.spark.sql.catalyst.expressions.Expression)
- object (class
org.apache.spark.sql.catalyst.encoders.ExpressionEncoder, class[])
- field (class:
org.apache.spark.sql.execution.aggregate.TypedAggregateExpression, name:
bEncoder, type: class
org.apache.spark.sql.catalyst.encoders.ExpressionEncoder)
- object (class
org.apache.spark.sql.execution.aggregate.TypedAggregateExpression,
AggKV(k#684,v#685))
- field (class:
org.apache.spark.sql.catalyst.expressions.aggregate.AggregateExpression,
name: aggregateFunction, type: class
org.apache.spark.sql.catalyst.expressions.aggregate.AggregateFunction)
- object (class
org.apache.spark.sql.catalyst.expressions.aggregate.AggregateExpression,
(AggKV(k#684,v#685),mode=Partial,isDistinct=false))
- element of array (index: 0)
- array (class [Ljava.lang.Object;, size 1)
- field (class: scala.collection.mutable.ArrayBuffer, name: array,
type: class [Ljava.lang.Object;)
- object (class scala.collection.mutable.ArrayBuffer,
ArrayBuffer((AggKV(k#684,v#685),mode=Partial,isDistinct=false)))
- field (class:
org.apache.spark.sql.execution.aggregate.SortBasedAggregate, name:
nonCompleteAggregateExpressions, type: interface scala.collection.Seq)
- object (class
org.apache.spark.sql.execution.aggregate.SortBasedAggregate,
SortBasedAggregate(key=[k#684],
functions=[(AggKV(k#684,v#685),mode=Partial,isDistinct=false)],
output=[k#684,value#731])
+- ConvertToSafe
+- Sort [k#684 ASC], false, 0
+- Project [k#684,v#685,k#684]
+- Scan ExistingRDD[k#684,v#685]
)
- field (class: org.apache.spark.sql.execution.ConvertToUnsafe,
name: child, type: class org.apache.spark.sql.execution.SparkPlan)
- object (class org.apache.spark.sql.execution.ConvertToUnsafe,
ConvertToUnsafe
+- SortBasedAggregate(key=[k#684],
functions=[(AggKV(k#684,v#685),mode=Partial,isDistinct=false)],
output=[k#684,value#731])
+- ConvertToSafe
+- Sort [k#684 ASC], false, 0
+- Project [k#684,v#685,k#684]
+- Scan ExistingRDD[k#684,v#685]
)
- field (class:
org.apache.spark.sql.execution.ConvertToUnsafe$$anonfun$1, name: $outer,
type: class org.apache.spark.sql.execution.ConvertToUnsafe)
- object (class
org.apache.spark.sql.execution.ConvertToUnsafe$$anonfun$1, <function1>)
at
org.apache.spark.serializer.SerializationDebugger$.improveException(SerializationDebugger.scala:40)
at
org.apache.spark.serializer.JavaSerializationStream.writeObject(JavaSerializer.scala:47)
at
org.apache.spark.serializer.JavaSerializerInstance.serialize(JavaSerializer.scala:101)
at
org.apache.spark.util.ClosureCleaner$.ensureSerializable(ClosureCleaner.scala:301)
... 134 more
Re: Dataset aggregateByKey equivalent
Posted by Michael Armbrust <mi...@databricks.com>.
Have you looked at aggregators?
https://docs.cloud.databricks.com/docs/spark/1.6/index.html#examples/Dataset%20Aggregator.html
On Fri, Apr 22, 2016 at 6:45 PM, Lee Becker <le...@hapara.com> wrote:
> Is there a way to do aggregateByKey on Datasets the way one can on an RDD?
>
> Consider the following RDD code to build a set of KeyVals into a DataFrame
> containing a column with the KeyVals' keys and a column containing lists of
> KeyVals. The end goal is to join it with collections which which will be
> similarly transformed.
>
> case class KeyVal(k: Int, v: Int)
>
>
> val keyVals = sc.parallelize(for (i <- 1 to 3; j <- 4 to 6) yield KeyVal(i,j))
>
> // function for appending to list
> val addToList = (s: List[KeyVal], v: KeyVal) => s :+ v
>
> // function for merging two lists
> val addLists = (s: List[KeyVal], t: List[KeyVal]) => s++t
>
> val keyAndKeyVals = keyVals.map(kv=> (kv.k, kv))
> val keyAndNestedKeyVals = keyAndKeyVals.
> aggregateByKey(List[KeyVal]())(addToList, addLists).
> toDF("key", "keyvals")
> keyAndNestedKeyVals.show
>
>
> which produces:
>
> +---+--------------------+
> |key| keyvals|
> +---+--------------------+
> | 1|[[1,4], [1,5], [1...|
> | 2|[[2,4], [2,5], [2...|
> | 3|[[3,4], [3,5], [3...|
> +---+--------------------+
>
> For a Dataset approach I tried the following to no avail:
>
> // Initialize as Dataset
> val keyVals = sc.parallelize(for (i <- 1 to 3; j <- 4 to 6) yield KeyVal(i,j)).
> toDS
>
> // Build key, keyVal mappings
> val keyValsByKey = keyVals.groupBy(kv=>(kv.k))
>
> case class NestedKeyVal(key: Int, keyvals: List[KeyVal])
>
> val convertToNested = (key: Int, keyValsIter: Iterator[KeyVal]) => NestedKeyVal(key=key, keyvals=keyValsIter.toList)
>
> val keyValsNestedByKey = keyValsByKey.mapGroups((key,keyvals) => convertToNested(key,keyvals))
> keyValsNestedByKey.show
>
>
> This and several other incantations using groupBy + mapGroups consistently
> gives me serialization problems. Is this because the iterator can not be
> guaranteed across boundaries?
> Or is there some issue with what a Dataset can encode in the interim.
> What other ways might I approach this problem?
>
> Thanks,
> Lee
>
>