Hello Michael, Sorry for the late replay .. I was crossing the world the last few days. I actually tried both ... REPEL and SparkApp. The reported exception was in App.
Unfortunately the data I have is not for distribution ... sorry about that. I saw it has been resolved.. I will try to reproduce the same error with dummy data. Thanks! On Thu, Jan 7, 2016 at 2:03 PM, Michael Armbrust <mich...@databricks.com> wrote: > Were you running in the REPL? > > On Thu, Jan 7, 2016 at 10:34 AM, Michael Armbrust <mich...@databricks.com> > wrote: > >> Thanks for providing a great description. I've opened >> https://issues.apache.org/jira/browse/SPARK-12696 >> >> I'm actually getting a different error (running in notebooks though). >> Something seems wrong either way. >> >>> >>> *P.S* mapping by name with case classes doesn't work if the order of >>> the fields of a case class doesn't match with the order of the DataFrame's >>> schema. >> >> >> We have tests for reordering >> <https://github.com/apache/spark/blob/master/sql/core/src/test/scala/org/apache/spark/sql/DatasetSuite.scala#L97> >> can >> you provide a smaller reproduction of this problem? >> >> On Wed, Jan 6, 2016 at 10:27 PM, Wail Alkowaileet <wael....@gmail.com> >> wrote: >> >>> Hey, >>> >>> I got an error when trying to map a Dataset df.as[CLASS] when I have >>> some nested case classes >>> I'm not sure if it's a bug ... or I did something wrong... or I missed >>> some configuration. >>> >>> >>> I did the following: >>> >>> *input snapshot* >>> >>> { >>> "count": "string", >>> "name": [{ >>> "addr_no": "string", >>> "dais_id": "string", >>> "display_name": "string", >>> "first_name": "string", >>> "full_name": "string", >>> "last_name": "string", >>> "r_id": "string", >>> "reprint": "string", >>> "role": "string", >>> "seq_no": "string", >>> "suffix": "string", >>> "wos_standard": "string" >>> }] >>> } >>> >>> *Case classes:* >>> >>> case class listType1(addr_no:String, dais_id:String, display_name:String, >>> first_name:String, full_name:String, last_name:String, r_id:String, >>> reprint:String, role:String, seq_no:String, suffix:String, >>> wos_standard:String) >>> case class DatasetType1(count:String, name:Array[listType1]) >>> >>> *Schema:* >>> root >>> |-- count: string (nullable = true) >>> |-- name: array (nullable = true) >>> | |-- element: struct (containsNull = true) >>> | | |-- addr_no: string (nullable = true) >>> | | |-- dais_id: string (nullable = true) >>> | | |-- display_name: string (nullable = true) >>> | | |-- first_name: string (nullable = true) >>> | | |-- full_name: string (nullable = true) >>> | | |-- last_name: string (nullable = true) >>> | | |-- r_id: string (nullable = true) >>> | | |-- reprint: string (nullable = true) >>> | | |-- role: string (nullable = true) >>> | | |-- seq_no: string (nullable = true) >>> | | |-- suffix: string (nullable = true) >>> | | |-- wos_standard: string (nullable = true) >>> >>> *Scala code:* >>> >>> import sqlContext.implicits._ >>> >>> val ds = df.as[DatasetType1] >>> >>> //Taking first() works fine >>> println(ds.first().count) >>> >>> //map() then first throws exception >>> println(ds.map(x => x.count).first()) >>> >>> >>> *Exception Message:* >>> Exception in thread "main" 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.SparkContext.runJob(SparkContext.scala:1857) >>> at org.apache.spark.SparkContext.runJob(SparkContext.scala:1929) >>> at org.apache.spark.rdd.RDD$$anonfun$collect$1.apply(RDD.scala:927) >>> 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.collect(RDD.scala:926) >>> at org.apache.spark.sql.Dataset.collect(Dataset.scala:668) >>> at main.main$.testAsterixRDDWithSparkSQL(main.scala:63) >>> at main.main$.main(main.scala:70) >>> at main.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 com.intellij.rt.execution.application.AppMain.main(AppMain.java:140) >>> Caused by: java.io.NotSerializableException: >>> scala.reflect.internal.Symbols$PackageClassSymbol >>> Serialization stack: >>> - object not serializable (class: >>> scala.reflect.internal.Symbols$PackageClassSymbol, value: package main) >>> - field (class: scala.reflect.internal.Types$ThisType, name: sym, type: >>> class scala.reflect.internal.Symbols$Symbol) >>> - object (class scala.reflect.internal.Types$UniqueThisType, main.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, >>> main.listType1) >>> - field (class: >>> org.apache.spark.sql.catalyst.ScalaReflection$$anonfun$org$apache$spark$sql$catalyst$ScalaReflection$$constructorFor$2, >>> name: elementType$1, type: class scala.reflect.api.Types$TypeApi) >>> - object (class >>> org.apache.spark.sql.catalyst.ScalaReflection$$anonfun$org$apache$spark$sql$catalyst$ScalaReflection$$constructorFor$2, >>> <function0>) >>> - field (class: >>> org.apache.spark.sql.catalyst.ScalaReflection$$anonfun$org$apache$spark$sql$catalyst$ScalaReflection$$constructorFor$2$$anonfun$apply$1, >>> name: $outer, type: class >>> org.apache.spark.sql.catalyst.ScalaReflection$$anonfun$org$apache$spark$sql$catalyst$ScalaReflection$$constructorFor$2) >>> - object (class >>> org.apache.spark.sql.catalyst.ScalaReflection$$anonfun$org$apache$spark$sql$catalyst$ScalaReflection$$constructorFor$2$$anonfun$apply$1, >>> <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>,cast(name#1 as >>> array<struct<display_name:string,first_name:string,full_name:string,reprint:string,role:string,wos_standard:string,last_name:string,dais_id:string,seq_no:string,suffix:string,r_id:string,addr_no:string>>),StructField(display_name,StringType,true),StructField(first_name,StringType,true),StructField(full_name,StringType,true),StructField(reprint,StringType,true),StructField(role,StringType,true),StructField(wos_standard,StringType,true),StructField(last_name,StringType,true),StructField(dais_id,StringType,true),StructField(seq_no,StringType,true),StructField(suffix,StringType,true),StructField(r_id,StringType,true),StructField(addr_no,StringType,true))) >>> - 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>,cast(name#1 as >>> array<struct<display_name:string,first_name:string,full_name:string,reprint:string,role:string,wos_standard:string,last_name:string,dais_id:string,seq_no:string,suffix:string,r_id:string,addr_no:string>>),StructField(display_name,StringType,true),StructField(first_name,StringType,true),StructField(full_name,StringType,true),StructField(reprint,StringType,true),StructField(role,StringType,true),StructField(wos_standard,StringType,true),StructField(last_name,StringType,true),StructField(dais_id,StringType,true),StructField(seq_no,StringType,true),StructField(suffix,StringType,true),StructField(r_id,StringType,true),StructField(addr_no,StringType,true)),array,ObjectType(class >>> [Lmain.listType1;))) >>> - writeObject data (class: scala.collection.immutable.$colon$colon) >>> - object (class scala.collection.immutable.$colon$colon, >>> List(invoke(count#0,toString,ObjectType(class java.lang.String)), >>> invoke(mapobjects(<function1>,cast(name#1 as >>> array<struct<display_name:string,first_name:string,full_name:string,reprint:string,role:string,wos_standard:string,last_name:string,dais_id:string,seq_no:string,suffix:string,r_id:string,addr_no:string>>),StructField(display_name,StringType,true),StructField(first_name,StringType,true),StructField(full_name,StringType,true),StructField(reprint,StringType,true),StructField(role,StringType,true),StructField(wos_standard,StringType,true),StructField(last_name,StringType,true),StructField(dais_id,StringType,true),StructField(seq_no,StringType,true),StructField(suffix,StringType,true),StructField(r_id,StringType,true),StructField(addr_no,StringType,true)),array,ObjectType(class >>> [Lmain.listType1;)))) >>> - field (class: org.apache.spark.sql.catalyst.expressions.NewInstance, >>> name: arguments, type: interface scala.collection.Seq) >>> - object (class org.apache.spark.sql.catalyst.expressions.NewInstance, >>> newinstance(class >>> main.DatasetType1,invoke(count#0,toString,ObjectType(class >>> java.lang.String)),invoke(mapobjects(<function1>,cast(name#1 as >>> array<struct<display_name:string,first_name:string,full_name:string,reprint:string,role:string,wos_standard:string,last_name:string,dais_id:string,seq_no:string,suffix:string,r_id:string,addr_no:string>>),StructField(display_name,StringType,true),StructField(first_name,StringType,true),StructField(full_name,StringType,true),StructField(reprint,StringType,true),StructField(role,StringType,true),StructField(wos_standard,StringType,true),StructField(last_name,StringType,true),StructField(dais_id,StringType,true),StructField(seq_no,StringType,true),StructField(suffix,StringType,true),StructField(r_id,StringType,true),StructField(addr_no,StringType,true)),array,ObjectType(class >>> [Lmain.listType1;)),false,ObjectType(class main.DatasetType1),None)) >>> - 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[count[0]: >>> string, name#ExprId(4,879d493d-efa1-4799-8fc4-5872ccb3b07b): >>> array<struct<display_name:string,first_name:string,full_name:string,reprint:string,role:string,wos_standard:string,last_name:string,dais_id:string,seq_no:string,suffix:string,r_id:string,addr_no:string>>]) >>> - field (class: org.apache.spark.sql.execution.MapPartitions, name: >>> tEncoder, type: class >>> org.apache.spark.sql.catalyst.encoders.ExpressionEncoder) >>> - object (class org.apache.spark.sql.execution.MapPartitions, >>> !MapPartitions <function1>, class[count[0]: string, >>> name#ExprId(4,879d493d-efa1-4799-8fc4-5872ccb3b07b): >>> array<struct<display_name:string,first_name:string,full_name:string,reprint:string,role:string,wos_standard:string,last_name:string,dais_id:string,seq_no:string,suffix:string,r_id:string,addr_no:string>>], >>> class[value[0]: string], [value#10] >>> +- ConvertToSafe >>> +- Scan JSONRelation[count#0,name#1] InputPaths: >>> ) >>> - field (class: org.apache.spark.sql.execution.MapPartitions$$anonfun$8, >>> name: $outer, type: class org.apache.spark.sql.execution.MapPartitions) >>> - object (class org.apache.spark.sql.execution.MapPartitions$$anonfun$8, >>> <function1>) >>> - field (class: >>> org.apache.spark.rdd.RDD$$anonfun$mapPartitionsInternal$1, name: f$22, >>> type: interface scala.Function1) >>> - object (class >>> org.apache.spark.rdd.RDD$$anonfun$mapPartitionsInternal$1, <function0>) >>> - field (class: >>> org.apache.spark.rdd.RDD$$anonfun$mapPartitionsInternal$1$$anonfun$apply$21, >>> name: $outer, type: class >>> org.apache.spark.rdd.RDD$$anonfun$mapPartitionsInternal$1) >>> - object (class >>> org.apache.spark.rdd.RDD$$anonfun$mapPartitionsInternal$1$$anonfun$apply$21, >>> <function3>) >>> - field (class: org.apache.spark.rdd.MapPartitionsRDD, name: f, type: >>> interface scala.Function3) >>> - object (class org.apache.spark.rdd.MapPartitionsRDD, >>> MapPartitionsRDD[6] at collect at main.scala:63) >>> - field (class: org.apache.spark.NarrowDependency, name: _rdd, type: >>> class org.apache.spark.rdd.RDD) >>> - object (class org.apache.spark.OneToOneDependency, >>> org.apache.spark.OneToOneDependency@7793b55d) >>> - writeObject data (class: scala.collection.immutable.$colon$colon) >>> - object (class scala.collection.immutable.$colon$colon, >>> List(org.apache.spark.OneToOneDependency@7793b55d)) >>> - field (class: org.apache.spark.rdd.RDD, name: >>> org$apache$spark$rdd$RDD$$dependencies_, type: interface >>> scala.collection.Seq) >>> - object (class org.apache.spark.rdd.MapPartitionsRDD, >>> MapPartitionsRDD[7] at collect at main.scala:63) >>> - field (class: org.apache.spark.rdd.RDD$$anonfun$collect$1, name: >>> $outer, type: class org.apache.spark.rdd.RDD) >>> - object (class org.apache.spark.rdd.RDD$$anonfun$collect$1, <function0>) >>> - field (class: org.apache.spark.rdd.RDD$$anonfun$collect$1$$anonfun$12, >>> name: $outer, type: class org.apache.spark.rdd.RDD$$anonfun$collect$1) >>> - object (class org.apache.spark.rdd.RDD$$anonfun$collect$1$$anonfun$12, >>> <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) >>> ... 19 more >>> >>> >>> >>> *P.S* mapping by name with case classes doesn't work if the order of >>> the fields of a case class doesn't match with the order of the DataFrame's >>> schema. >>> >>> >>> -- >>> >>> *Regards,* >>> Wail Alkowaileet >>> >> >> > -- *Regards,* Wail Alkowaileet
