Spark Streaming Custom Receivers
Spark Streaming can receive streaming data from any arbitrary data source beyond the ones for which it has built-in support (that is, beyond Kafka, Kinesis, files, sockets, etc.). This requires the developer to implement a receiver that is customized for receiving data from the concerned data source. This guide walks through the process of implementing a custom receiver and using it in a Spark Streaming application. Note that custom receivers can be implemented in Scala or Java.
Implementing a Custom Receiver
This starts with implementing a Receiver (Scala doc, Java doc). A custom receiver must extend this abstract class by implementing two methods
onStart(): Things to do to start receiving data.
onStop(): Things to do to stop receiving data.
onStop() must not block indefinitely. Typically,
onStart() would start the threads
that are responsible for receiving the data, and
onStop() would ensure that these threads receiving the data
are stopped. The receiving threads can also use
Receiver method, to check whether they
should stop receiving data.
Once the data is received, that data can be stored inside Spark
store(data), which is a method provided by the Receiver class.
There are a number of flavors of
store() which allow one to store the received data
record-at-a-time or as whole collection of objects / serialized bytes. Note that the flavor of
store() used to implement a receiver affects its reliability and fault-tolerance semantics.
This is discussed later in more detail.
Any exception in the receiving threads should be caught and handled properly to avoid silent
failures of the receiver.
restart(<exception>) will restart the receiver by
onStop() and then calling
onStart() after a delay.
stop(<exception>) will call
onStop() and terminate the receiver. Also,
reports an error message to the driver (visible in the logs and UI) without stopping / restarting
The following is a custom receiver that receives a stream of text over a socket. It treats ‘\n’ delimited lines in the text stream as records and stores them with Spark. If the receiving thread has any error connecting or receiving, the receiver is restarted to make another attempt to connect.
Using the custom receiver in a Spark Streaming application
The custom receiver can be used in a Spark Streaming application by using
streamingContext.receiverStream(<instance of custom receiver>). This will create
an input DStream using data received by the instance of custom receiver, as shown below:
The full source code is in the example CustomReceiver.scala.
The full source code is in the example JavaCustomReceiver.java.
As discussed in brief in the Spark Streaming Programming Guide, there are two kinds of receivers based on their reliability and fault-tolerance semantics.
- Reliable Receiver - For reliable sources that allow sent data to be acknowledged, a reliable receiver correctly acknowledges to the source that the data has been received and stored in Spark reliably (that is, replicated successfully). Usually, implementing this receiver involves careful consideration of the semantics of source acknowledgements.
- Unreliable Receiver - An unreliable receiver does not send acknowledgement to a source. This can be used for sources that do not support acknowledgement, or even for reliable sources when one does not want or need to go into the complexity of acknowledgement.
To implement a reliable receiver, you have to use
store(multiple-records) to store data.
This flavor of
store is a blocking call which returns only after all the given records have
been stored inside Spark. If the receiver’s configured storage level uses replication
(enabled by default), then this call returns after replication has completed.
Thus it ensures that the data is reliably stored, and the receiver can now acknowledge the
source appropriately. This ensures that no data is lost when the receiver fails in the middle
of replicating data – the buffered data will not be acknowledged and hence will be later resent
by the source.
An unreliable receiver does not have to implement any of this logic. It can simply receive
records from the source and insert them one-at-a-time using
store(single-record). While it does
not get the reliability guarantees of
store(multiple-records), it has the following advantages:
- The system takes care of chunking that data into appropriate sized blocks (look for block interval in the Spark Streaming Programming Guide).
- The system takes care of controlling the receiving rates if the rate limits have been specified.
- Because of these two, unreliable receivers are simpler to implement than reliable receivers.
The following table summarizes the characteristics of both types of receivers
Simple to implement.
System takes care of block generation and rate control. No fault-tolerance guarantees, can lose data on receiver failure.
Strong fault-tolerance guarantees, can ensure zero data loss.
Block generation and rate control to be handled by the receiver implementation.
Implementation complexity depends on the acknowledgement mechanisms of the source.