gistfile1.txt

package org.apache.spark.examples.streaming

import org.apache.spark.SparkConf
import org.apache.spark.streaming._

/**
 * Counts words cumulatively in UTF8 encoded, '\n' delimited text received from the network every
 * second starting with initial value of word count.
 * Usage: StatefulNetworkWordCount <hostname> <port>
 *   <hostname> and <port> describe the TCP server that Spark Streaming would connect to receive
 *   data.
 *
 * To run this on your local machine, you need to first run a Netcat server
 *    `$ nc -lk 9999`
 * and then run the example
 *    `$ bin/run-example
 *      org.apache.spark.examples.streaming.StatefulNetworkWordCount localhost 9999`
 */
object StatefulNetworkWordCount {
  
  def checkpointDir = "file:/tmp/spark/checkpoint/test"
  
  def main(args: Array[String]) {
    if (args.length < 2) {
      System.err.println("Usage: StatefulNetworkWordCount <hostname> <port>")
      System.exit(1)
    }

    val Array(hostname, port) = args

    //StreamingExamples.setStreamingLogLevels()

    val ssc =  StreamingContext.getOrCreate(checkpointDir, () => createContext(hostname, port.toInt))

    ssc.start()
    ssc.awaitTermination()
  }
    
  def createContext(hostname: String, port: Int) : StreamingContext = {   

    val sparkConf = new SparkConf().setAppName("StatefulNetworkWordCount")
    sparkConf.set("spark.streaming.receiver.writeAheadLog.enable", "true")
    sparkConf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
    
    // Create the context with a 10 second batch size
    val ssc = new StreamingContext(sparkConf, Seconds(10))
    ssc.checkpoint(checkpointDir)

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

    // Update the cumulative count using mapWithState
    // This will give a DStream made of state (which is the cumulative count of the words)
    val mappingFunc = (word: String, one: Option[Int], state: State[Int]) => {
      val sum = one.getOrElse(0) + state.getOption.getOrElse(0)
      val output = (word, sum)
      state.update(sum)
      output
    }

    wordDstream.mapWithState(StateSpec.function(mappingFunc) ).
      map(x => (x._1.toUpperCase(), x._2)).
      foreachRDD(rdd => {      
          rdd.foreachPartition(partition => {
            partition.foreach(println)
          })
    })

    ssc
  }
}