How to use endless Scala streams as a source in Spark Streaming?

Question

Suppose I really want Stream.from(0) as InputDStream . How can i do this? The only way I can see is to use StreamingContext#queueStream , but I need to either insert elements from another stream or Queue subclass to create a queue that behaves like an infinite stream, both of which feel hacked.

What is the right way to do this?

Answer 1

I do not think that it is available in Spark by default, but it is easy to implement it using ReceiverInputDStream.

 import org.apache.spark.storage.StorageLevel import org.apache.spark.streaming.StreamingContext import org.apache.spark.streaming.dstream.ReceiverInputDStream import org.apache.spark.streaming.receiver.Receiver class InfiniteStreamInputDStream[T]( @transient ssc_ : StreamingContext, stream: Stream[T], storageLevel: StorageLevel ) extends ReceiverInputDStream[T](ssc_) { override def getReceiver(): Receiver[T] = { new InfiniteStreamReceiver(stream, storageLevel) } } class InfiniteStreamReceiver[T](stream: Stream[T], storageLevel: StorageLevel) extends Receiver[T](storageLevel) { // Stateful iterator private val streamIterator = stream.iterator private class ReadAndStore extends Runnable { def run(): Unit = { while (streamIterator.hasNext) { val next = streamIterator.next() store(next) } } } override def onStart(): Unit = { new Thread(new ReadAndStore).run() } override def onStop(): Unit = { } } 

Answer 2

The slightly modified tat code works with Spark 2.0:

 import org.apache.spark.storage.StorageLevel import org.apache.spark.streaming.StreamingContext import org.apache.spark.streaming.dstream.ReceiverInputDStream import org.apache.spark.streaming.receiver.Receiver import scala.reflect.ClassTag class InfiniteDStream[T: ClassTag]( @transient ssc_ : StreamingContext, stream: Stream[T], storageLevel: StorageLevel ) extends ReceiverInputDStream[T](ssc_) { override def getReceiver(): Receiver[T] = { new InfiniteStreamReceiver(stream, storageLevel) } } class InfiniteStreamReceiver[T](stream: Stream[T], storageLevel: StorageLevel) extends Receiver[T](storageLevel) { private class ReadAndStore extends Runnable { def run(): Unit = { stream.foreach(store) } } override def onStart(): Unit = { val t = new Thread(new ReadAndStore) t.setDaemon(true) t.run() } override def onStop(): Unit = {} }