Found this a bug in spark 1.4.0: SPARK-8368
<https://issues.apache.org/jira/browse/SPARK-8368>
Thanks!
Terry
On Thu, Jul 2, 2015 at 1:20 PM, Terry Hole <hujie.ea...@gmail.com> wrote:
> All,
>
> I am using spark console 1.4.0 to do some tests, when a create a newly
> HiveContext (Line 18 in the code) in my test function, it always throw
> exception like below (It works in spark console 1.3.0), but if i removed
> the HiveContext (The line 18 in the code) in my function, it works fine.
> Any idea what's wrong with this?
>
> java.lang.ClassNotFoundException:
> $iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$
> iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$anonfun$streamingTest$1$$anonfun$apply$1
> at java.net.URLClassLoader.findClass(URLClassLoader.java:381) at
> java.lang.ClassLoader.loadClass(ClassLoader.java:424)
> at sun.misc.Launcher$AppClassLoader.loadClass(Launcher.java:331)
> at java.lang.ClassLoader.loadClass(ClassLoader.java:357)
> at java.lang.Class.forName0(Native Method)
> at java.lang.Class.forName(Class.java:348)
> at
> org.apache.spark.util.InnerClosureFinder$$anon$4.visitMethodInsn(Clos
> ureCleaner.scala:455)
> at
> com.esotericsoftware.reflectasm.shaded.org.objectweb.asm.ClassReader.accept(Unknown
> Source)
> at
> com.esotericsoftware.reflectasm.shaded.org.objectweb.asm.ClassReader.accept(Unknown
> Source)
> at
> org.apache.spark.util.ClosureCleaner$.getInnerClosureClasses(ClosureCleaner.scala:101)
> at
> org.apache.spark.util.ClosureCleaner$.org$apache$spark$util$ClosureCleaner$$clean(ClosureCleaner.scala:197)
> at
> org.apache.spark.util.ClosureCleaner$.clean(ClosureCleaner.scala:132)
> at org.apache.spark.SparkContext.clean(SparkContext.scala:1891)
> at
> org.apache.spark.streaming.dstream.DStream$$anonfun$foreachRDD$1.apply$mcV$sp(DStream.scala:630)
> at
> org.apache.spark.streaming.dstream.DStream$$anonfun$foreachRDD$1.apply(DStream.scala:629)
> at
> org.apache.spark.streaming.dstream.DStream$$anonfun$foreachRDD$1.apply(DStream.scala:629)
> at
> org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:148)
> at
> org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:109)
> at org.apache.spark.SparkContext.withScope(SparkContext.scala:681)
> at
> org.apache.spark.streaming.StreamingContext.withScope(StreamingContext.scala:258)
> at
> org.apache.spark.streaming.dstream.DStream.foreachRDD(DStream.scala:629)
> at
> $iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC
> $$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC.streamingTest(<console>:98)
> at
> $iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC
> $$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC.<init>(<console>:93)
> at
> $iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC
> $$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC.<init>(<console>:98)
>
>
>
> 1 import org.apache.spark._ 2 import org.apache.spark.SparkContext._ 3
> import org.apache.spark.streaming.{ StreamingContext, Seconds, Minutes, Time
> } 4 import org.apache.spark.streaming.StreamingContext._ 5 import
> org.apache.spark.rdd.RDD 6 import org.apache.spark.streaming.dstream.DStream
> 7 import org.apache.spark.HashPartitioner 8 import
> org.apache.spark.storage.StorageLevel 9 import org.apache.spark.sql._10
> import org.apache.spark.sql.hive._11 import
> scala.collection.mutable.{Queue}12 import scala.concurrent.Future13 import
> scala.concurrent.ExecutionContext.Implicits.global14 15 def
> streamingTest(args: Array[String]) {16 println(">>> create
> streamingContext.")17 val ssc = new StreamingContext(sc, Seconds(1))18
> *val sqlContext2 = new HiveContext(sc)*19 20 val accum =
> sc.accumulator(0, "End Accumulator")21 val queue =
> scala.collection.mutable.Queue(sc.textFile("G:/pipe/source"))22 val
> textSource = ssc.queueStream(queue, true)23 textSource.foreachRDD(rdd =>
> { rdd.foreach( item => {accum += 1} ) })24 textSource.foreachRDD(rdd =>
> {25 var sample = rdd.take(10)26 if
> (sample.length > 0) {27 sample.foreach(item =>
> println("#= " + item))28 }29 })30
> println(">>> Start streaming context.")31 ssc.start()32 val stopFunc
> = Future {var isRun = true; var duration = 0; while (isRun) {
> Thread.sleep(1000); duration += 1; if ( accum.value > 0 || duration >= 120)
> {println("### STOP SSC ###");ssc.stop(false, true); duration = 0; isRun =
> false} }}33 ssc.awaitTermination()34 println(">>> Streaming context
> terminated.")35 }36 37 streamingTest(null)38
>
> Thanks
> Terry
>
