Streaming: add value only stream and update readme

Author: sunheehnus

README.md

@@ -5,6 +5,8 @@ A library for reading and writing data from and to [Redis](http://redis.io) with
 
 Spark-Redis provides access to all of Redis' data structures - String, Hash, List, Set and Sorted Set - from Spark as RDDs. The library can be used both with Redis stand-alone as well as clustered databases. When used with Redis cluster, Spark-Redis is aware of its partitioning scheme and adjusts in response to resharding and node failure events.
 
+Spark-Redis also provides Spark-Streaming support.
+
 ## Minimal requirements
 You'll need the the following to use Spark-Redis:
 
@@ -15,7 +17,7 @@ You'll need the the following to use Spark-Redis:
 
 ## Known limitations
 
-* Java, Python and R API bindings are not provided at this time 
+* Java, Python and R API bindings are not provided at this time
 * The package was only tested with the following stack:
  - Apache Spark v1.4.0
  - Scala v2.10.4
@@ -72,13 +74,13 @@ import com.redislabs.provider.redis._
 sc = new SparkContext(new SparkConf()
       .setMaster("local")
       .setAppName("myApp")
-      
+
       // initial redis host - can be any node in cluster mode
       .set("redis.host", "localhost")
-      
+
       // initial redis port
       .set("redis.port", "6379")
-      
+
       // optional redis AUTH password
       .set("redis.auth", "")
   )
@@ -248,6 +250,18 @@ sc.toRedisZSET(zsetRDD, zsetName)
 
 The above example demonstrates storing data in Redis in a Sorted Set. The `zsetRDD` in the example should contain pairs of members and their scores, whereas `zsetName` is the name for that key.
 
+### Streaming
+Spark-Redis support streaming data from Redis instance/cluster, currently streaming data are fetched from Redis' List by the `blpop` command. Users are required to provide an array which stores all the List names they are interested in. The [storageLevel](http://spark.apache.org/docs/latest/streaming-programming-guide.html#data-serialization) is `MEMORY_AND_DISK_SER_2` by default, you can change it on your demand. The stream is a `(listName, value)` stream by default, but if you don't care about which list feeds the value, set the `withStreamName` as `false` and you will get the only `value` stream.
+```
+import org.apache.spark.streaming.{Seconds, StreamingContext}
+import org.apache.spark.storage.StorageLevel
+import com.redislabs.provider.redis._
+val ssc = new StreamingContext(sc, Seconds(1))
+val redisStream = ssc.createRedisStream(Array("foo", "bar"), storageLevel = StorageLevel.MEMORY_AND_DISK_2, withStreamName = true)
+```
+The above example will get a (listName, value) stream from `foo` list and `bar` list
+
+
 ### Connecting to Multiple Redis Clusters/Instances
 
 ```

src/main/scala/com/redislabs/provider/redis/redisFunctions.scala

@@ -385,13 +385,31 @@ object RedisContext extends Serializable {
   }
 }
 
+/**
+  * RedisStreamingContext extends StreamingContext's functionality with Redis
+  *
+  * @param ssc a spark StreamingContext
+  */
 class RedisStreamingContext(@transient val ssc: StreamingContext) extends Serializable {
-  def createRedisStream(keys: Array[String], storageLevel: StorageLevel)
+  /**
+    * @param keys an Array[String] which consists all the Lists we want to listen to
+    * @param storageLevel the receiver' storage tragedy of received data, default as MEMORY_AND_DISK_2
+    * @param withStreamName if it is true the streaming will be (listName, value),
+    *                       else the streaming will be only (value)
+    *                       default as true
+    */
+  def createRedisStream(keys: Array[String],
+                        storageLevel: StorageLevel = StorageLevel.MEMORY_AND_DISK_2,
+                        withStreamName: Boolean = true)
                        (implicit redisConfig: RedisConfig = new RedisConfig(new
                            RedisEndpoint(ssc.sparkContext.getConf))) = {
-    new RedisInputDStream(ssc, keys, redisConfig, storageLevel)
+    withStreamName match {
+      case true => new RedisInputDStream(ssc, keys, storageLevel, redisConfig, classOf[(String, String)])
+      case false => new RedisInputDStream(ssc, keys, storageLevel, redisConfig, classOf[String])
+    }
   }
 }
+
 trait RedisFunctions {
   implicit def toRedisContext(sc: SparkContext): RedisContext = new RedisContext(sc)
   implicit def toRedisStreamingContext(ssc: StreamingContext): RedisStreamingContext = new RedisStreamingContext(ssc)

src/main/scala/com/redislabs/provider/redis/streaming/RedisInputDStream.scala

@@ -1,56 +1,65 @@
 package com.redislabs.provider.redis.streaming
 
-import com.redislabs.provider.redis.RedisConfig
 import org.apache.curator.utils.ThreadUtils
 import org.apache.spark.storage.StorageLevel
 import org.apache.spark.streaming.StreamingContext
 import org.apache.spark.streaming.receiver.Receiver
 import org.apache.spark.streaming.dstream.ReceiverInputDStream
-
 import redis.clients.jedis._
-
+import scala.reflect.{ClassTag, classTag}
 import scala.util.control.NonFatal
+import com.redislabs.provider.redis.RedisConfig
 
-class RedisInputDStream(_ssc: StreamingContext,
-                        keys: Array[String],
-                        redisConfig: RedisConfig,
-                        storageLevel: StorageLevel)
-  extends ReceiverInputDStream[(String, String)](_ssc) {
-  def getReceiver(): Receiver[(String, String)] = {
-    new RedisReceiver(keys, storageLevel, redisConfig)
+class RedisInputDStream[T: ClassTag](_ssc: StreamingContext,
+                                     keys: Array[String],
+                                     storageLevel: StorageLevel,
+                                     redisConfig: RedisConfig,
+                                     streamType: Class[T])
+  extends ReceiverInputDStream[T](_ssc) {
+  def getReceiver(): Receiver[T] = {
+    new RedisReceiver(keys, storageLevel, redisConfig, streamType)
   }
 }
 
-private class RedisReceiver(keys: Array[String],
-                            storageLevel: StorageLevel,
-                            redisConfig: RedisConfig)
-  extends Receiver[(String, String)](storageLevel) {
+
+private class RedisReceiver[T: ClassTag](keys: Array[String],
+                                         storageLevel: StorageLevel,
+                                         redisConfig: RedisConfig,
+                                         streamType: Class[T])
+  extends Receiver[T](storageLevel) {
 
   def onStart() {
-    val executorPool = ThreadUtils.newFixedThreadPool(keys.length, "BlockLists Streamming")
+    val executorPool = ThreadUtils.newFixedThreadPool(keys.length, "BlockLists Streaming")
     try {
+      /* start a executor for each interested List */
       keys.foreach{ key =>
         executorPool.submit(new MessageHandler(redisConfig.connectionForKey(key), key))
       }
     } finally {
       executorPool.shutdown()
     }
-
   }
 
   def onStop() {
   }
 
   private class MessageHandler(conn: Jedis, key: String) extends Runnable {
     def run() {
-      val keys: Array[String] = Array(key)
       try {
         while(!isStopped) {
-          val res = conn.blpop(Integer.MAX_VALUE, keys:_*)
-          res match {
-            case null => "Really?"
-            case _ => store((res.get(0), res.get(1)))
+          val res = {
+            val response = conn.blpop(0, key)
+            if (classTag[T] == classTag[String]) {
+              response.get(1)
+            }
+            else if (classTag[T] == classTag[(String, String)]) {
+              (response.get(0), response.get(1))
+            }
+            else {
+              throw new scala.Exception("Unknown Redis Streaming type")
+            }
           }
+          store(res.asInstanceOf[T])
         }
       } catch {
         case NonFatal(e) =>
@@ -61,4 +70,3 @@ private class RedisReceiver(keys: Array[String],
     }
   }
 }
-