Yes, you need to follow the documentation. Configure your stream, including the transformations made to it, inside the getOrCreate function.
On Tue, Jul 28, 2015 at 3:14 AM, Guillermo Ortiz <konstt2...@gmail.com> wrote: > I'm using SparkStreaming and I want to configure checkpoint to manage > fault-tolerance. > I've been reading the documentation. Is it necessary to create and > configure the InputDSStream in the getOrCreate function? > > I checked the example in > https://github.com/apache/spark/blob/master/examples/src/main/scala/org/apache/spark/examples/streaming/RecoverableNetworkWordCount.scala > and it looks like it does everything inside of the function. Should I put > all the logic of the application inside on it?? I think that that's not the > way... > > If I just create the context I got an error: > Exception in thread "main" org.apache.spark.SparkException: > org.apache.spark.streaming.kafka.DirectKafkaInputDStream@1e12a5a6 has not > been initialized > at > org.apache.spark.streaming.dstream.DStream.isTimeValid(DStream.scala:266) > at > org.apache.spark.streaming.dstream.InputDStream.isTimeValid(InputDStream.scala:51) > > > I'm not pretty good with Scala.. the code that I did > def functionToCreateContext(): StreamingContext = { > val sparkConf = new > SparkConf().setMaster("local[2]").setAppName("app") > val ssc = new StreamingContext(sparkConf, Seconds(5)) // new context > > ssc.checkpoint("/tmp/spark/metricsCheckpoint") // set checkpoint > directory > ssc > } > > > val ssc = StreamingContext.getOrCreate("/tmp/spark/metricsCheckpoint", > functionToCreateContext _) > val kafkaParams = Map[String, String]("metadata.broker.list" -> > args(0)) > val topics = args(1).split("\\,") > val directKafkaStream = KafkaUtils.createDirectStream[String, String, > StringDecoder, StringDecoder](ssc, kafkaParams, topics.toSet) > > directKafkaStream.foreachRDD { rdd => ... > >
