Uploaded image for project: 'Spark'
  1. Spark
  2. SPARK-13390

Java Spark createDataFrame with List parameter bug

    XMLWordPrintableJSON

Details

    • Bug
    • Status: Resolved
    • Minor
    • Resolution: Fixed
    • 1.6.0
    • 1.6.1
    • Spark Core
    • None

    • Java spark, Linux

    Description

      I noticed the following bug while testing the dataframe SQL join capabilities.
      Instructions to reproduce it:

      • Read a text file from local file system using JavaSparkContext#texFile method
      • Create a list of related custom objects based on the previously created JavaRDD, using the map function
      • Create a dataframe using SQLContext createDataFrame(java.util.List, Class) method
      • Count the dataframe elements using dataframe#count method

      It crashes with the following stacktrace error:
      org.apache.spark.sql.catalyst.errors.package$TreeNodeException: execute, tree:
      TungstenAggregate(key=[], functions=[(count(1),mode=Final,isDistinct=false)], output=count#7L)
      +- TungstenExchange SinglePartition, None
      +- TungstenAggregate(key=[], functions=[(count(1),mode=Partial,isDistinct=false)], output=count#10L)
      +- LocalTableScan [[empty row],[empty row],[empty row],[empty row],[empty row],[empty row],[empty row],[empty row],[empty row],[empty row]]

      at org.apache.spark.sql.catalyst.errors.package$.attachTree(package.scala:49)
      at org.apache.spark.sql.execution.aggregate.TungstenAggregate.doExecute(TungstenAggregate.scala:80)
      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.executeCollect(SparkPlan.scala:166)
      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$count$1.apply(DataFrame.scala:1554)
      at org.apache.spark.sql.DataFrame$$anonfun$count$1.apply(DataFrame.scala:1553)
      at org.apache.spark.sql.DataFrame.withCallback(DataFrame.scala:2138)
      at org.apache.spark.sql.DataFrame.count(DataFrame.scala:1553)
      at injection.EMATests.joinTest1(EMATests.java:259)
      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:498)
      at org.junit.runners.model.FrameworkMethod$1.runReflectiveCall(FrameworkMethod.java:50)
      at org.junit.internal.runners.model.ReflectiveCallable.run(ReflectiveCallable.java:12)
      at org.junit.runners.model.FrameworkMethod.invokeExplosively(FrameworkMethod.java:47)
      at org.junit.internal.runners.statements.InvokeMethod.evaluate(InvokeMethod.java:17)
      at org.junit.runners.ParentRunner.runLeaf(ParentRunner.java:325)
      at org.junit.runners.BlockJUnit4ClassRunner.runChild(BlockJUnit4ClassRunner.java:78)
      at org.junit.runners.BlockJUnit4ClassRunner.runChild(BlockJUnit4ClassRunner.java:57)
      at org.junit.runners.ParentRunner$3.run(ParentRunner.java:290)
      at org.junit.runners.ParentRunner$1.schedule(ParentRunner.java:71)
      at org.junit.runners.ParentRunner.runChildren(ParentRunner.java:288)
      at org.junit.runners.ParentRunner.access$000(ParentRunner.java:58)
      at org.junit.runners.ParentRunner$2.evaluate(ParentRunner.java:268)
      at org.junit.internal.runners.statements.RunBefores.evaluate(RunBefores.java:26)
      at org.junit.internal.runners.statements.RunAfters.evaluate(RunAfters.java:27)
      at org.junit.runners.ParentRunner.run(ParentRunner.java:363)
      at org.junit.runner.JUnitCore.run(JUnitCore.java:137)
      at com.intellij.junit4.JUnit4IdeaTestRunner.startRunnerWithArgs(JUnit4IdeaTestRunner.java:69)
      at com.intellij.rt.execution.junit.JUnitStarter.prepareStreamsAndStart(JUnitStarter.java:234)
      at com.intellij.rt.execution.junit.JUnitStarter.main(JUnitStarter.java:74)
      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:498)
      at com.intellij.rt.execution.application.AppMain.main(AppMain.java:144)
      Caused by: org.apache.spark.sql.catalyst.errors.package$TreeNodeException: execute, tree:
      TungstenExchange SinglePartition, None
      +- TungstenAggregate(key=[], functions=[(count(1),mode=Partial,isDistinct=false)], output=count#10L)
      +- LocalTableScan [[empty row],[empty row],[empty row],[empty row],[empty row],[empty row],[empty row],[empty row],[empty row],[empty row]]

      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.aggregate.TungstenAggregate$$anonfun$doExecute$1.apply(TungstenAggregate.scala:86)
      at org.apache.spark.sql.execution.aggregate.TungstenAggregate$$anonfun$doExecute$1.apply(TungstenAggregate.scala:80)
      at org.apache.spark.sql.catalyst.errors.package$.attachTree(package.scala:48)
      ... 46 more
      Caused by: org.apache.spark.sql.catalyst.errors.package$TreeNodeException: execute, tree:
      TungstenAggregate(key=[], functions=[(count(1),mode=Partial,isDistinct=false)], output=count#10L)
      +- LocalTableScan [[empty row],[empty row],[empty row],[empty row],[empty row],[empty row],[empty row],[empty row],[empty row],[empty row]]

      at org.apache.spark.sql.catalyst.errors.package$.attachTree(package.scala:49)
      at org.apache.spark.sql.execution.aggregate.TungstenAggregate.doExecute(TungstenAggregate.scala:80)
      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)
      ... 54 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.aggregate.TungstenAggregate$$anonfun$doExecute$1.apply(TungstenAggregate.scala:86)
      at org.apache.spark.sql.execution.aggregate.TungstenAggregate$$anonfun$doExecute$1.apply(TungstenAggregate.scala:80)
      at org.apache.spark.sql.catalyst.errors.package$.attachTree(package.scala:48)
      ... 63 more
      Caused by: java.io.NotSerializableException: scala.collection.Iterator$$anon$11
      Serialization stack:

      • object not serializable (class: scala.collection.Iterator$$anon$11, value: empty iterator)
      • field (class: scala.collection.Iterator$$anonfun$toStream$1, name: $outer, type: interface scala.collection.Iterator)
      • object (class scala.collection.Iterator$$anonfun$toStream$1, <function0>)
      • field (class: scala.collection.immutable.Stream$Cons, name: tl, type: interface scala.Function0)
      • object (class scala.collection.immutable.Stream$Cons, Stream([TRI1,N,TNW,160000,0006093430000,E,2016-02-01-15.20.31.434000], [TRI2,N,TNW,170000,0006093430000,E,2016-02-01-15.20.31.434000], [TRI3,N,TNW,180000,0006093430000,E,2016-02-01-15.20.31.434000], [TRI4,N,TNW,190000,0006093430000,E,2016-02-01-15.20.31.434000], [CHI1,N,TNY,200000,0006093430000,E,2016-02-01-15.20.31.434000], [CHI2,N,TNY,210000,0006093430000,E,2016-02-01-15.20.31.434000], [CHI3,N,TNY,220000,0006093430000,E,2016-02-01-15.20.31.434000], [CHI4,N,TNY,230000,0006093430000,E,2016-02-01-15.20.31.434000], [CRU1,N,TNY,240000,0006093430000,E,2016-02-01-15.20.31.434000], [CRU2,N,TNY,250000,0006093430000,E,2016-02-01-15.20.31.434000]))
      • field (class: scala.collection.immutable.Stream$$anonfun$map$1, name: $outer, type: class scala.collection.immutable.Stream)
      • object (class scala.collection.immutable.Stream$$anonfun$map$1, <function0>)
      • field (class: scala.collection.immutable.Stream$Cons, name: tl, type: interface scala.Function0)
      • object (class scala.collection.immutable.Stream$Cons, Stream([empty row], [empty row], [empty row], [empty row], [empty row], [empty row], [empty row], [empty row], [empty row], [empty row]))
      • field (class: org.apache.spark.sql.execution.LocalTableScan, name: rows, type: interface scala.collection.Seq)
      • object (class org.apache.spark.sql.execution.LocalTableScan, LocalTableScan [[empty row],[empty row],[empty row],[empty row],[empty row],[empty row],[empty row],[empty row],[empty row],[empty row]]
        )
      • field (class: org.apache.spark.sql.execution.aggregate.TungstenAggregate, name: child, type: class org.apache.spark.sql.execution.SparkPlan)
      • object (class org.apache.spark.sql.execution.aggregate.TungstenAggregate, TungstenAggregate(key=[], functions=[(count(1),mode=Partial,isDistinct=false)], output=count#10L)
        +- LocalTableScan [[empty row],[empty row],[empty row],[empty row],[empty row],[empty row],[empty row],[empty row],[empty row],[empty row]]
        )
      • field (class: org.apache.spark.sql.execution.aggregate.TungstenAggregate$$anonfun$doExecute$1, name: $outer, type: class org.apache.spark.sql.execution.aggregate.TungstenAggregate)
      • object (class org.apache.spark.sql.execution.aggregate.TungstenAggregate$$anonfun$doExecute$1, <function0>)
      • field (class: org.apache.spark.sql.execution.aggregate.TungstenAggregate$$anonfun$doExecute$1$$anonfun$2, name: $outer, type: class org.apache.spark.sql.execution.aggregate.TungstenAggregate$$anonfun$doExecute$1)
      • object (class org.apache.spark.sql.execution.aggregate.TungstenAggregate$$anonfun$doExecute$1$$anonfun$2, <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)
        ... 75 more

      A workaround is to use create dataframe directly on JavaRDDs instead of lists

      Attachments

        Activity

          People

            zsxwing Shixiong Zhu
            niemaz mike niemaz
            Votes:
            0 Vote for this issue
            Watchers:
            2 Start watching this issue

            Dates

              Created:
              Updated:
              Resolved: