The Receiver API

The most fundamental receiver consists of three methods:

def onStart()

Starts the reception of data from the external sources. In practice, onStart should asynchronously initiate the inbound data collection process and return immediately.

def store(...)

Delivers one or several data elements to Spark Streaming. store must be called from the asynchronous process initiated by onStart whenever there is new data available.

def stop(...)

Stops the receiving process. stop must take care of properly cleaning up any resources used by the receiving process started by onStart.

This model provides a generic interface that can be implemented to integrate a wide range of streaming data providers. Note how the genericity abstracts over the data delivery method of the streaming system. We can implement an always-connected push-based receiver, like a TCP client socket, as well as a request-based pull connector, like a REST/HTTP connector to some system.

How Receivers Work

The task of the receiver is to collect data from the stream source and deliver it to Spark Streaming. Intuitively, it’s very easy to understand: as data comes in, it’s collected and packaged in blocks during the time of the batch interval. As soon as each batch interval period of time is completed, the collected data blocks are given to Spark for processing.

Figure 18-3 depicts how the timing of this sequence of events takes place. At the start of the streaming process, the receiver begins collecting data. At the end of the t0 interval, the first collected block #0 is given to Spark for processing. At time t2, Spark is processing the data block collected at t1, while the receiver is collecting the data corresponding to block #2.

Figure 18-3. Receiver in action

We can generalize that, at any point in time, Spark is processing the previous batch of data, while the receivers are collecting data for the current interval. After a batch has been processed by Spark (like #0 in Figure 18-3), it can be cleaned up. The time when that RDD will be cleaned up is determined by the spark.cleaner.ttl setting.

The Receiver’s Data Flow

Figure 18-4 illustrates the data flow of a Spark application in this case.

Figure 18-4. The data flow of Spark’s receiver

In this figure, we can see that data ingestion occurs as a job that is translated into a single task on one executor. This task deals with connecting to a data source and initiates the data transfer. It is managed from the Spark Context, a bookkeeping object that resides within the driver machine.

On each block interval tick (as measured on the executor that is running the receiver), this machine groups the data received for the previous block interval into a block. The block is then registered with the Block Manager, also present in the bookkeeping of the Spark Context, on the driver. This process initiates the replication of the data that this block represents, to ensure that the source data in Spark Streaming is replicated the number of times indicated by the storage level.

On each batch interval tick (as measured on the driver), the driver groups the data received for the previous batch interval, which has been replicated the correct number of times, into an RDD. Spark registers this RDD with the JobScheduler, initiating the scheduling of a job on that particular RDD—in fact, the whole point of Spark Streaming’s microbatching model consists of repeatedly scheduling the user-defined program on successive batched RDDs of data.

The Internal Data Resilience

The fact that receivers are independent jobs has consequences, in particular for the resource usage and data delivery semantics. To execute its data collection job, the receiver consumes one core on an executor regardless of the amount of work it needs to do. Therefore, using a single streaming receiver will result in data ingestion done in sequence by a single core in an executor, which becomes the limiting factor to the amount of data that Spark Streaming can ingest.

The base unit of replication for Spark is a block: any block can be on one or several machines (up to the persistence level indicated in the configuration, so at most two by default), and it is only when a block has reached that persistence level that it can be processed. Therefore, it is only when an RDD has every block replicated that it can be taken into account for job scheduling.

At the Spark engine side, each block becomes a partition of the RDD. The combination of a partition of data and the work that needs to be applied to it becomes a task. Each task can be processed in parallel, usually on the executor that contains the data locally. Therefore, the level of parallelism we can expect for an RDD obtained from a single receiver is exactly the ratio of the batch interval to the block interval, as defined in Equation 18-1.

In Spark, the usual rule of thumb for task parallelism is to have two to three times ratio of the number of tasks to the number of available executor cores. Taking a factor of three for our discussion, we should set the block interval to that shown in Equation 18-2.

Receiver Parallelism

We mentioned that a single receiver would be a limiting factor to the amount of data that we can process with Spark Streaming.

A simple way to increase incoming data throughput is to declare more DStreams at the code level. Each DStream will be attached to its own consumer—and therefore each will have its own core consumption—on the cluster. The DStream operation union allows us to merge those streams, ensuring that we produce a single pipeline of data from our various input streams.

Let’s assume that we create DStreams in parallel an put them in a sequence:

val inputDstreams: Seq[DStream[(K,V)]] = Seq.fill(parallelism: Int) {
... // the input Stream creation function
}
val joinedStream  = ssc.union(inputDstreams)

In this way, we can exploit the receiver parallelism, here represented by the #parallelism factor of concurrently created DStreams.

Balancing Resources: Receivers Versus Processing Cores

Given that each created receiver consumes its own core in a cluster, increasing consumer parallelism has consequences in the number of cores available for processing in our cluster.

Let’s imagine that we have a 12-core cluster that we want to dedicate to our streaming analytics application. When using a single receiver, we use one core for receiving and nine cores for processing data. The cluster might be underutilized because a single receiver might not receive enough data to give work to all of the available processing cores. Figure 18-5 illustrates that situation, in which the green nodes are being used for processing, and the gray nodes remain idle.

Figure 18-5. Single-receiver allocation

To improve cluster utilization, we increase the number of receivers, as we just discussed. In our hypothetical scenario, using four receivers provides us with a much better resource allocation, as shown in Figure 18-6.

Figure 18-6. Multiple-receiver allocation

The batch interval is fixed by the needs of the analysis and remains the same. What should the block interval be? Well, four DStreams ingesting in parallel will necessarily create four times as many blocks per block interval as one single DStream would. Therefore, with the same block interval, the number of partitions of the unionized DStream will be four times what it was in the original case. Hence, we can’t use the same block interval. Instead, we should use the following:

Because we want at least three partitions, we round this number down to the nearest millisecond.

Generalizing, with an arbitrary set of characteristics, we should use this:

Where the total number of cores used in the system is as follows:

Achieving Zero Data Loss with the Write-Ahead Log

The original receiver design, previous to Spark v1.2, had a major design flaw. While the receiver is collecting data for the current block, that data is only found in a memory buffer in the receiver process. Only when the block is completed and delivered does it become replicated in the cluster. In case of failure of the receiver, the data in that buffer gets lost and cannot be recovered, causing data loss.

To prevent data loss, data collected by the receiver is additionally appended to a log file on a reliable filesystem. This is known as the write-ahead log (WAL), a component commonly used in database design to guarantee reliable and durable data reception.

The WAL is an append-only structure in which data is written before it is delivered for processing. When data is known to be correctly processed, its entry in the log is marked as processed. In the database world, the equivalent process is the commit of the transaction in which the data is involved, making this log also known as the commit log.

In case of failure, data from the WAL is replayed from the record following the last registered commit, compensating in that way for the potential loss of data of the receiver. The combination of the WAL and the receiver is known as the reliable receiver. Streaming sources based on the reliable receiver model are known as reliable sources.