Spark Streaming Cache and Transformations

Question

I am new to spark, I am using Spark streaming with Kafka..

My streaming duration is 1 second.

Assume i get 100 records in 1st batch and 120 records in 2nd batch and 80 records in 3rd batch

--> {sec 1   1,2,...100} --> {sec 2 1,2..120} --> {sec 3 1,2,..80}

I apply my logic in 1st batch and have a result => result1

I want to use result1 while processing 2nd batch and have a combined result of both result1 and 120 records of 2nd batch as => result2

I tried to cache the result but I am not able to get the cached result1 in 2s is it possible? or show some light on how to achieve my goal here?

 JavaPairReceiverInputDStream<String, String> messages =   KafkaUtils.createStream(jssc, String.class,String.class, StringDecoder.class,StringDecoder.class, kafkaParams, topicMap, StorageLevel.MEMORY_AND_DISK_SER_2());

I process messages and find word which is the result for 1 second.

if(resultCp!=null){
                resultCp.print();
                result = resultCp.union(words.mapValues(new Sum()));

            }else{
                result = words.mapValues(new Sum());
            }

 resultCp =  result.cache();

when in 2nd batch the resultCp should not be null but it returns null value so at any given time i have that particular seconds data alone i want to find the cumulative result. Do any one know how to do it..

I learnt that once spark streaming is started jssc.start() the control is no more at our end it lies with spark. So is it possible to send the result of 1st batch to 2nd batch to find the accumulated value?

Any help is much much appreciated. Thanks in advance.

maasg maasg · Accepted Answer · 2014-10-20T10:00:20

I think you are looking for updateStateByKey which creates a new DStream by applying a cummulative function to the provided DStream and some state. This example from the Spark example package covers the case in the question:

First, you need an update function that takes the new values and the previously known value:

val updateFunc = (values: Seq[Int], state: Option[Int]) => {
  val currentCount = values.sum

  val previousCount = state.getOrElse(0)

  Some(currentCount + previousCount)
}

That function is used to create a Dstream that cummulates values from a source dstream. Like this:

// Create a NetworkInputDStream on target ip:port and count the
// words in input stream of \n delimited test (eg. generated by 'nc')
val lines = ssc.socketTextStream(args(0), args(1).toInt)
val words = lines.flatMap(_.split(" "))
val wordDstream = words.map(x => (x, 1))

// Update the cumulative count using updateStateByKey
// This will give a Dstream made of state (which is the cumulative count of the words)
val stateDstream = wordDstream.updateStateByKey[Int](updateFunc)

Source: https://github.com/apache/spark/blob/master/examples/src/main/scala/org/apache/spark/examples/streaming/StatefulNetworkWordCount.scala

Spark Streaming Cache and Transformations

1 Answers