Re: kafka + Spark Streaming with checkPointing fails to start with

[Alexander Krasheninnikov <a....@corp.badoo.com>]

I had same problem.
The solution, I've found was to use:
JavaStreamingContext streamingContext = 
JavaStreamingContext.getOrCreate('checkpoint_dir', contextFactory);

ALL configuration should be performed inside contextFactory. If you try 
to configure streamContext after ::getOrCreate, you receive an error 
"has not been initialized".

On 13.05.2015 00:51, Ankur Chauhan wrote:
> Hi,
>
> I have a simple application which fails with the following exception only when the application is restarted (i.e. the checkpointDir has entires from a previous execution):
>
> Exception in thread "main" org.apache.spark.SparkException: org.apache.spark.streaming.dstream.ShuffledDStream@2264e43c has not been initialized
> 	at org.apache.spark.streaming.dstream.DStream.isTimeValid(DStream.scala:266)
> 	at org.apache.spark.streaming.dstream.DStream$$anonfun$getOrCompute$1.apply(DStream.scala:287)
> 	at org.apache.spark.streaming.dstream.DStream$$anonfun$getOrCompute$1.apply(DStream.scala:287)
> 	at scala.Option.orElse(Option.scala:257)
> 	at org.apache.spark.streaming.dstream.DStream.getOrCompute(DStream.scala:284)
> 	at org.apache.spark.streaming.dstream.ForEachDStream.generateJob(ForEachDStream.scala:38)
> 	at org.apache.spark.streaming.DStreamGraph$$anonfun$1.apply(DStreamGraph.scala:116)
> 	at org.apache.spark.streaming.DStreamGraph$$anonfun$1.apply(DStreamGraph.scala:116)
> 	at scala.collection.TraversableLike$$anonfun$flatMap$1.apply(TraversableLike.scala:251)
> 	at scala.collection.TraversableLike$$anonfun$flatMap$1.apply(TraversableLike.scala:251)
> 	at scala.collection.mutable.ResizableArray$class.foreach(ResizableArray.scala:59)
> 	at scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:47)
> 	at scala.collection.TraversableLike$class.flatMap(TraversableLike.scala:251)
> 	at scala.collection.AbstractTraversable.flatMap(Traversable.scala:105)
> 	at org.apache.spark.streaming.DStreamGraph.generateJobs(DStreamGraph.scala:116)
> 	at org.apache.spark.streaming.scheduler.JobGenerator$$anonfun$restart$4.apply(JobGenerator.scala:227)
> 	at org.apache.spark.streaming.scheduler.JobGenerator$$anonfun$restart$4.apply(JobGenerator.scala:222)
> 	at scala.collection.IndexedSeqOptimized$class.foreach(IndexedSeqOptimized.scala:33)
> 	at scala.collection.mutable.ArrayOps$ofRef.foreach(ArrayOps.scala:108)
> 	at org.apache.spark.streaming.scheduler.JobGenerator.restart(JobGenerator.scala:222)
> 	at org.apache.spark.streaming.scheduler.JobGenerator.start(JobGenerator.scala:90)
> 	at org.apache.spark.streaming.scheduler.JobScheduler.start(JobScheduler.scala:67)
> 	at org.apache.spark.streaming.StreamingContext.start(StreamingContext.scala:512)
> 	at com.brightcove.analytics.tacoma.RawLogProcessor$.start(RawLogProcessor.scala:115)
> 	at com.brightcove.analytics.tacoma.Main$delayedInit$body.apply(Main.scala:15)
> 	at scala.Function0$class.apply$mcV$sp(Function0.scala:40)
> 	at scala.runtime.AbstractFunction0.apply$mcV$sp(AbstractFunction0.scala:12)
> 	at scala.App$$anonfun$main$1.apply(App.scala:71)
> 	at scala.App$$anonfun$main$1.apply(App.scala:71)
> 	at scala.collection.immutable.List.foreach(List.scala:318)
> 	at scala.collection.generic.TraversableForwarder$class.foreach(TraversableForwarder.scala:32)
> 	at scala.App$class.main(App.scala:71)
> 	at com.brightcove.analytics.tacoma.Main$.main(Main.scala:5)
> 	at com.brightcove.analytics.tacoma.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:483)
> 	at org.apache.spark.deploy.SparkSubmit$.org$apache$spark$deploy$SparkSubmit$$runMain(SparkSubmit.scala:569)
> 	at org.apache.spark.deploy.SparkSubmit$.doRunMain$1(SparkSubmit.scala:166)
> 	at org.apache.spark.deploy.SparkSubmit$.submit(SparkSubmit.scala:189)
> 	at org.apache.spark.deploy.SparkSubmit$.main(SparkSubmit.scala:110)
> 	at org.apache.spark.deploy.SparkSubmit.main(SparkSubmit.scala)
>
> The relavant source is:
>
> class RawLogProcessor(ssc: StreamingContext, topic: String, kafkaParams: Map[String, String]) {
>    // create kafka stream
>    val rawlogDStream = KafkaUtils.createDirectStream[String, Object, StringDecoder, KafkaAvroDecoder](ssc, kafkaParams, Set(topic))
>    //KafkaUtils.createStream[String, Object, StringDecoder, KafkaAvroDecoder](ssc, kafkaParams, Map("qa-rawlogs" -> 10), StorageLevel.MEMORY_AND_DISK_2)
>
>    val eventStream = rawlogDStream
>      .map({
>        case (key, rawlogVal) =>
>          val record = rawlogVal.asInstanceOf[GenericData.Record]
>          val rlog = RawLog.newBuilder()
>            .setId(record.get("id").asInstanceOf[String])
>            .setAccount(record.get("account").asInstanceOf[String])
>            .setEvent(record.get("event").asInstanceOf[String])
>            .setTimestamp(record.get("timestamp").asInstanceOf[Long])
>            .setUserAgent(record.get("user_agent").asInstanceOf[String])
>            .setParams(record.get("params").asInstanceOf[java.util.Map[String, String]])
>            .build()
>          val norm = Normalizer(rlog)
>          (key, rlog.getEvent, norm)
>      })
>
>    val videoViewStream = eventStream
>      .filter(_._2 == "video_view")
>      .filter(_._3.isDefined)
>      .map((z) => (z._1, z._3.get))
>      .map((z) => (z._1, z._2.asInstanceOf[VideoView]))
>      .cache()
>
>    // repartition by (deviceType, DeviceOS)
>    val deviceTypeVideoViews = videoViewStream.map((v) => ((v._2.getDeviceType, v._2.getDeviceOs), 1))
>      .reduceByKeyAndWindow(_ + _, Durations.seconds(10))
>      .print()
> }
>
> object RawLogProcessor extends Logging {
>
>    /**
>     * If str is surrounded by quotes it return the content between the quotes
>     */
>    def unquote(str: String) = {
>      if (str != null && str.length >= 2 && str.charAt(0) == '\"' && str.charAt(str.length - 1) == '\"')
>        str.substring(1, str.length - 1)
>      else
>        str
>    }
>
>    val checkpointDir = "/tmp/checkpointDir_tacoma"
>    var sparkConfig: Config = _
>    var ssc: StreamingContext = _
>    var processor: Option[RawLogProcessor] = None
>
>    val createContext: () => StreamingContext = () => {
>      val batchDurationSecs = sparkConfig.getDuration("streaming.batch_duration", TimeUnit.SECONDS)
>      val sparkConf = new SparkConf()
>      sparkConf.registerKryoClasses(Array(classOf[VideoView], classOf[RawLog], classOf[VideoEngagement], classOf[VideoImpression]))
>      sparkConfig.entrySet.asScala
>        .map(kv => kv.getKey -> kv.getValue)
>        .foreach {
>          case (k, v) =>
>            val value = unquote(v.render())
>
>            logInfo(s"spark.$k = $value")
>
>            sparkConf.set(s"spark.$k", value)
>        }
>
>      // calculate sparkContext and streamingContext
>      new StreamingContext(sparkConf, Durations.seconds(batchDurationSecs))
>    }
>
>    def createProcessor(sparkConf: Config, kafkaConf: Config): RawLogProcessor = {
>      sparkConfig = sparkConf
>      ssc = StreamingContext.getOrCreate(checkpointPath = checkpointDir, creatingFunc = createContext, createOnError = true)
>      ssc.checkpoint(checkpointDir)
>      // kafkaProperties
>      val kafkaParams = kafkaConf.entrySet.asScala
>        .map(kv => kv.getKey -> unquote(kv.getValue.render()))
>        .toMap
>
>      logInfo(s"Initializing kafkaParams = $kafkaParams")
>      // create processor
>      new RawLogProcessor(ssc, kafkaConf.getString("rawlog.topic"), kafkaParams)
>    }
>
>    def apply(sparkConfig: Config, kafkaConf: Config) = {
>      if (processor.isEmpty) {
>        processor = Some(createProcessor(sparkConfig, kafkaConf))
>      }
>      processor.get
>    }
>
>    def start() = {
>      ssc.start()
>      ssc.awaitTermination()
>    }
>
> }
>
> Extended logs: https://gist.githubusercontent.com/ankurcha/f35df63f0d8a99da0be4/raw/ec96b932540ac87577e4ce8385d26699c1a7d05e/spark-console.log
>
> Could someone tell me what it causes this problem? I tried looking at the stacktrace but I am not very familiar with the codebase to make solid assertions.
> Any ideas as to what may be happening here.
>
> --- Ankur Chauhan