Monday, May 2, 2016

Sliding window using : reduceByKeyAndWindow()

This example demonstrates the Sliding window using the Transformation reduceByKeyAndWindow(). This Python stub is used to provide Input to this Application

Input

Input for the Streaming Application is denoted by this color

python3.5 ./a21_stub.py inputfile=./c15_input2.yaml 
Press ENTER when Ready...
[  0]
     [.9] => data=lion || secondscounter=0.9
[  1]
     [.8] => data=lion || secondscounter=1.8
[  2]
     [.7] => data=lion || secondscounter=2.7
[  3]
     [.6] => data=lion || secondscounter=3.6
[  4]
     [.5] => data=lion || secondscounter=4.5
[  5]
     [.4] => data=lion || secondscounter=5.4
[  6]
     [.4] => data=lion || secondscounter=6.4
[  7]
     [.2] => data=lion || secondscounter=7.2
[  8]
     [.1] => data=lion || secondscounter=8.1
[  9]
     [.0] => data=lion || secondscounter=9.0
[ 10]
     [.0] => data=lion || secondscounter=10.0
     [.8] => data=lion || secondscounter=10.8
[ 11]
     [.7] => data=lion || secondscounter=11.7
[ 12]
     [.6] => data=lion || secondscounter=12.6

Output

(lion,(2,List(0.9, 1.8)))
(lion,(4,List(0.9, 1.8, 2.7, 3.6)))
(lion,(6,List(0.9, 1.8, 2.7, 3.6, 4.5, 5.4)))
(lion,(7,List(2.7, 3.6, 4.5, 5.4, 6.4, 7.2, 8.1)))
(lion,(7,List(4.5, 5.4, 6.4, 7.2, 8.1, 9.0, 10.0)))
(lion,(7,List(6.4, 7.2, 8.1, 9.0, 10.0, 10.8, 11.7)))
(lion,(5,List(9.0, 10.0, 10.8, 11.7, 12.6)))
(lion,(3,List(10.8, 11.7, 12.6)))
(lion,(1,List(12.6)))

Source Code

  1. package a11_TCP_IP.a15_Window
  2.  
  3. import org.apache.spark.SparkConf
  4. import org.apache.spark.SparkContext
  5. import org.apache.spark.streaming.Seconds
  6. import org.apache.spark.streaming.StreamingContext
  7. import java.text.SimpleDateFormat
  8. import java.util.Calendar
  9.  
  10. /*
  11. spark-submit --master local[*] --class a11_TCP_IP.a15_Window.a11_WindowExample --driver-java-options -XX:MaxPermSize=300m target/scala-2.10/sparkstreamingexamples_2.10-1.0.jar
  12. */
  13. object a11_WindowExample {
  14.   
  15.   def main(args: Array[String]) {
  16.     
  17.     val conf = new SparkConf().setAppName("Window_Example")
  18.     val sc = new SparkContext(conf)
  19.  
  20.     // Set Batch Interval
  21.     val stc = new StreamingContext(sc, Seconds(2))
  22.     
  23.     // Create an Input stream that receives data from
  24.     // TCP/IP Socket
  25.     val linesDstream = stc.socketTextStream("localhost", 9999)
  26.  
  27.     val splittedDs = linesDstream.map{ x =>
  28.       x.split("\\|\\|")
  29.        .map(_.trim)
  30.     }
  31.  
  32.     val filteredListDs = splittedDs.map{ x =>
  33.       val retval = for { element <- x
  34.           val keyNval = element.split("=")
  35.           // Ensure key & value are both present
  36.           if keyNval.size >= 2
  37.         } yield {
  38.           val splitted = element.split("=")
  39.           // Create Tuple of Key and Value
  40.           splitted(0) -> splitted(1)
  41.         }
  42.       retval
  43.     }
  44.     
  45.     // Convert the Splitted tokens into a Map 
  46.     val dnryListDs = filteredListDs.map{ x => x.toMap }
  47.     
  48.     val keyvaluepairsDs = dnryListDs.map { x =>
  49.       val data = x.getOrElse("data", "")
  50.       (data, x)
  51.     }
  52.  
  53.     val wordsDs = keyvaluepairsDs.flatMap{ x =>
  54.       val xsplitted = x._1.split(" ").map(_.trim)
  55.       val wordNmeta = for { element <- xsplitted
  56.         } yield {
  57.           (element, (  1, 
  58.                        List((x._2.getOrElse("secondscounter", ""))
  59.                     )))
  60.         }
  61.       wordNmeta  
  62.     }
  63.     
  64.     def reduceFn(  x:(Int, List[String]), y:(Int, List[String])  ) 
  65.                                   : (Int, List[String]) = {
  66.       val sum = x._1 + y._1
  67.       val timelist = x._2 ::: y._2
  68.       (sum, timelist)
  69.     }
  70.     
  71.     // Sliding window with Window duration of 6 seconds and
  72.     // sliding duration of 2 seconds
  73.     // reduceFn _ : Is a Partially Applied Function
  74.     //            : Denotes, all the arguments are passed @ runtime
  75.     val reducedDs = wordsDs.reduceByKeyAndWindow(reduceFn _, 
  76.                                         Seconds(6), Seconds(2))
  77.     
  78.     // Apply the given function to each RDD in this DStream
  79.     reducedDs.foreachRDD{ rdd =>
  80.        val minNSec = getMinNsec()
  81.        //println(f"Window at $minNSec%s")
  82.        //println("-----------------------")
  83.        rdd.collect().foreach(println)
  84.        //println("")
  85.     }
  86.     
  87.     // Start the Computation & wait for it to terminate
  88.     stc.start()
  89.     stc.awaitTermination()
  90.   }
  91.   
  92.   def getMinNsec() : String = {
  93.     val now = Calendar.getInstance().getTime()
  94.     val myFormat = new SimpleDateFormat("mm.ss.SS")
  95.     val minNmillisec = myFormat.format(now)
  96.     val splitted = minNmillisec.split("\\.")
  97.     val min = splitted(0)
  98.     val sec = splitted(1)
  99.     val millisec = splitted(2)
  100.     val millisecTruncated = millisec(0)
  101.     f"[$min%2s][$sec%2s.$millisecTruncated%s]"
  102.   }
  103. }