diff options
author | Prashant Sharma <prashant.iiith@gmail.com> | 2013-01-19 21:20:49 +0530 |
---|---|---|
committer | Prashant Sharma <prashant.iiith@gmail.com> | 2013-01-19 22:04:07 +0530 |
commit | 56b9bd197c522f33e354c2e9ad7e76440cf817e9 (patch) | |
tree | bf62af73fa0272db1ae6b7c01a6d0d0307a85be8 /docs | |
parent | bb6ab92e31b7aad464cf8262bc3567fdeb4c14c4 (diff) | |
download | spark-56b9bd197c522f33e354c2e9ad7e76440cf817e9.tar.gz spark-56b9bd197c522f33e354c2e9ad7e76440cf817e9.tar.bz2 spark-56b9bd197c522f33e354c2e9ad7e76440cf817e9.zip |
Plug in actor as stream receiver API
Diffstat (limited to 'docs')
-rw-r--r-- | docs/plugin-custom-receiver.md | 101 |
1 files changed, 101 insertions, 0 deletions
diff --git a/docs/plugin-custom-receiver.md b/docs/plugin-custom-receiver.md new file mode 100644 index 0000000000..41e6a17e2c --- /dev/null +++ b/docs/plugin-custom-receiver.md @@ -0,0 +1,101 @@ +--- +layout: global +title: Tutorial - Spark streaming, Plugging in a custom receiver. +--- + +A "Spark streaming" receiver can be a simple network stream, streams of messages from a message queue, files etc. A receiver can also assume roles more than just receiving data like filtering, preprocessing, to name a few of the possibilities. The api to plug-in any user defined custom receiver is thus provided to encourage development of receivers which may be well suited to ones specific need. + +This guide shows the programming model and features by walking through a simple sample receiver and corresponding Spark Streaming application. + + +## A quick and naive walk-through + +### Write a simple receiver + +This starts with implementing [Actor](#References) + +Following is a simple socket text-stream receiver, which is appearently overly simplified using Akka's socket.io api. + +{% highlight scala %} + + class SocketTextStreamReceiver (host:String, + port:Int, + bytesToString: ByteString => String) extends Actor { + + override def preStart = IOManager(context.system).connect(host, port) + + def receive = { + case IO.Read(socket, bytes) => context.parent ! Data(bytesToString(bytes)) + } + } + + +{% endhighlight %} + + +_Please see implementations of NetworkReceiver for more generic NetworkReceivers._ + +### A sample spark application + +* First create a Spark streaming context with master url and batchduration. + +{% highlight scala %} + + val ssc = new StreamingContext(master, "WordCountCustomStreamSource", + Seconds(batchDuration)) + +{% endhighlight %} + +* Plug-in the actor configuration into the spark streaming context and create a DStream. + +{% highlight scala %} + + val lines = ssc.pluggableActorStream[String](Props(new SocketTextStreamReceiver( + "localhost",8445, z => z.utf8String)),"SocketReceiver") + +{% endhighlight %} + +* Process it. + +{% highlight scala %} + + val words = lines.flatMap(_.split(" ")) + val wordCounts = words.map(x => (x, 1)).reduceByKey(_ + _) + + wordCounts.print() + ssc.start() + + +{% endhighlight %} + +* After processing it, stream can be tested using the netcat utility. + + $ nc -l localhost 8445 + hello world + hello hello + + +## Multiple homogeneous/heterogeneous receivers. + +A DStream union operation is provided for taking union on multiple input streams. + +{% highlight scala %} + + val lines = ssc.pluggableActorStream[String](Props(new SocketTextStreamReceiver( + "localhost",8445, z => z.utf8String)),"SocketReceiver") + + // Another socket stream receiver + val lines2 = ssc.pluggableActorStream[String](Props(new SocketTextStreamReceiver( + "localhost",8446, z => z.utf8String)),"SocketReceiver") + + val union = lines.union(lines2) + +{% endhighlight %} + +Above stream can be easily process as described earlier. + +_A more comprehensive example is provided in the spark streaming examples_ + +## References + +1.[Akka Actor documentation](http://doc.akka.io/docs/akka/2.0.5/scala/actors.html) |