The Performance Balance of Spark Streaming

Performance tuning in Spark Streaming can sometimes be complex, but it always begins with the simple equilibrium between the batch interval and the batch processing time. We can view the batch processing time as the time cost we have to complete the processing of all the received data and any other related bookkeeping, whereas the batch interval is the budget we have allocated. Much like the financial analogy, a healthy application will fit its processing cost within the allocated budget. Although it might happen that in some particular moments when the pressure goes up, we go beyond the budget, we must see that in the longer term, our balance is preserved. An application that exceeds this time-budget balance over a long period will result in a systemic failure, usually resulting in the crash of the application due to resource exhaustion.

External Factors that Influence the Job’s Performance

Finally, if all those factors have been taken into account and you are still witnessing spikes in the processing delay of your jobs, another aspect that we really need to pay attention to is the changing conditions on the cluster.

For example, other systems colocated on our cluster may impact our shared processing resources: the Hadoop Distributed File System (HDFS) is known to have had bugs in its older versions that constrained concurrent disk writes.¹ Therefore, we might be running a cluster at a very stable rate, while simultaneously, a different job—that might not even be Spark related—can require heavy use of the disk. This can affect the following:

• Data ingestion in the reliable receiver model, when using a write-ahead log (WAL)

• Checkpointing time

• Actions of our stream processing that involve saving data to the disk

To alleviate this issue of external impacts on our job through disk usage, we could do the following:

• Use a distributed in-memory cache such as Alluxio²

• Reduce disk pressure by saving structured, small data in a NoSQL database rather than on files

• Avoid colocating more disk-intensive applications with Spark than is strictly necessary

Disk access is only one of the possible bottlenecks that could affect our job through resource sharing with the cluster. Another possibility can be network starvation or, more generally, the existence of workloads that cannot be monitored and scheduled through our resource manager.

How to Improve Performance?

In the previous section, we discussed the intrinsic and extrinsic factors that can influence the performance of a Spark Streaming job.

Let’s imagine that we are in a situation in which we developed a job and we observe certain issues that affect the performance and, hence, the stability of the job. The first step to take would be to gain insights in the different performance indicators of our job, perhaps using the techniques outlined in “Understanding Job Performance Using the Streaming UI”.

We use that information as a comparison baseline as well as guidance to use one or more of the different strategies that follow.

Tweaking the Batch Interval

A strategy that is mentioned frequently is to lengthen the batch interval. This approach might help to improve some parallelism and resource usage issues. For example, if we increase a batch interval from one minute to five minutes, we have to serialize only the tasks that are the components of our job once every five minutes instead of once every minute—a five-fold reduction.

Nonetheless, the batches of our stream will represent five minutes’ worth of data seen “over the wire” instead of one, and because most of the instability issues are caused by an inadequate distribution of our resources to the throughput of our stream, the batch interval might change little to this imbalance. More important the batch interval that we seek to implement is often of high semantic value in our analysis; if only because, as we have seen in Chapter 21, it constrains the windowing and sliding intervals that we can create on an aggregated stream. Changing these analysis semantics to accommodate processing constraints should be envisioned only as a last resort.

A more compelling strategy consists of reducing general inefficiencies, such as using a fast serialization library or implementing algorithms with better performance characteristics. We can also accelerate disk-writing speeds by augmenting or replacing our distributed filesystem with an in-memory cache, such as Alluxio. When that’s not sufficient, we should consider adding more resources to our cluster, letting us distribute the stream on more executors by correspondingly augmenting the number of partitions that we use through, for example, block interval tuning.

Limiting the Data Ingress with Fixed-Rate Throttling

If getting more resources is absolutely impossible, we need to look at reducing the number of data elements that we must deal with.

Since version 1.3, Spark includes a fixed-rate throttling that allows it to accept a maximum number of elements. We can set this by adding spark.streaming.receiver.maxRate to a value in elements per second in your Spark configuration. Note that for the receiver-based consumers, this limitation is enforced at block creation and simply refuses to read any more elements from the data source if the throttle limit has been reached.

For the Kafka direct connector, there’s a dedicated configuration spark.streaming.kafka.maxRatePerPartition that sets the max rate limit per partition in the topic in records per second. When using this option, be mindful that the total rate will be as follows:

Note that this behavior does not, in and of itself, include any signaling; Spark will just let a limited amount of elements, and pick up the reading of new elements on the next batch interval. This has consequences on the system that is feeding data into Spark:

• If this is a pull-based system, such as in Kafka, Flume, and others, the input system could compute the number of elements read and manage the overflow data in a custom fashion.

• If the input system is more prosaically a buffer (file buffer, TCP buffer), it will overflow after a few block intervals (because our stream has a large throughput than the throttle) and will periodically be flushed (deleted) when this happens.

As a consequence, throttled ingestion in Spark can exhibit some “jitter” in the elements read, because Spark reads every element until an underlying TCP or file buffer, used as a queue for “late” elements, reaches capacity and is flushed as a whole. The effect of this is that the input stream is separated in large intervals of processed elements interspersed with “holes” (dropped elements) of a regular size (e.g., one TCP buffer).

Backpressure

The queue-based system we have described with fixed-rate throttling has the disadvantage that it makes it obscure for our entire pipeline to understand where inefficiencies lie. Indeed, we have considered a data source (e.g., a TCP socket) that consists of reading data from an external server (e.g., an HTTP server), into a local system-level queue (a TCP buffer), before Spark feeds this data in an application-level buffer (Spark Streaming’s RDDs). Unless we use a listener tied to our Spark Streaming receiver, it is challenging to detect and diagnose that our system is congested and, if so, where the congestion occurs.

The external server could perhaps decide, if it was aware that our Spark Streaming cluster is congested, to react on that signal and use its own approach to either delay or select the incoming elements to Spark. More important, it could make the congestion information flow back up the stream to the data producers it depends on, calling every part of the pipeline to be aware of and help with the congestion. It would also allow any monitoring system to have a better view of how and where congestion happens in our system helping with resource management and tuning.

The upstream-flowing, quantified signal about congestion is called backpressure. This is a continuous signaling that explicitly says how many elements the system in question (here, our Spark Streaming cluster) can be expected to process at this specific instant. Backpressure signaling has an advantage with respect to throttling because it is set up as a dynamic signal that varies in function to the influx of elements and the state of the queue in Spark. As such, it does not affect the system if there is no congestion, and it does not require tuning of an arbitrary limit, avoiding the associated risks in misconfiguration (underused resources if the limit is too restrictive; overflow if the limit is too permissive).

This approach has been available in Spark since version 1.5 and can, in a nutshell, provide dynamic throttling.

Dynamic Throttling

In Spark Streaming, dynamic throttling is regulated by default with a Proportional-Integral-Derivative (PID) controller, which observes an error signal as the difference between the latest ingestion rate, observed on a batch interval in terms of number of elements per second, and the processing rate, which is the number of elements that have been processed per second. We could consider this error as the imbalance between the number of elements coming in and the number of elements going out of Spark at the current instant (with an “instant” rounded to a full batch interval).

The PID controller then aims at regulating the number of ingested elements on the next batch interval by taking into account the following:

• A proportional term (the error at this instant)

• An integral or “historic” term (the sum of all errors in the past; here, the number of unprocessed elements lying in the queue)

• A derivative or “speed” term (the rate at which the number of elements has been diminishing in the past)

The PID then attempts to compute an ideal number depending on these three factors.

Backpressure-based dynamic throttling in Spark can be turned on by setting spark.streaming.backpressure.enabled to true in your Spark configuration. Another variable spark.streaming.backpressure.initialRate dictates the number of elements per second the throttling should initially expect. You should set it slightly above your best estimate of the throughput of your stream to allow the algorithm to “warm up.”

spark.streaming.backpressure.pid.proportional
spark.streaming.backpressure.pid.integral
spark.streaming.backpressure.pid.derived

  def compute(
      time: Long,
      elements: Long,
      processingDelay: Long,
      schedulingDelay: Long): Option[Double]
}

Caching

Caching in Spark Streaming is a feature that, when well manipulated, can significantly speed up the computation performed by your application. This seems to be counterintuitive given that the base RDDs representing the data stored in the input of a computation are actually replicated twice before any job runs on them.

However, over the lifetime of your application, there might be a very long pipeline that takes your computation from those base RDDs to some very refined and structured representations of the data that usually involves a key–value tuple. At the end of the computation performed by your application, you are probably looking at doing some distribution of the output of your computation into various outlets: data stores or databases such as Cassandra, for example. That distribution usually involves looking at the data computed during the previous batch interval and finding which portions of the output data should go where.

A typical use case for that is to look at the keys in the RDD of the structured output data (the last DStream in your computation), to find exactly where to put the results of your computation outside of Spark, depending on these keys. Another use case would be to look for only some specific elements on the RDD received during the last batch. Indeed, your RDD might actually be the output of a computation that depends not only on the last batch of data, but on many prior events received since the start of the application. The last step of your pipeline might summarize the state of your system. Looking at that RDD of output structured results, we might be searching for some elements that pass certain criteria, comparing new results with previous values, or distributing data to different organizational entities, to name a few cases.

For example, think of anomaly detection. You might compute some metrics or features on values (users or elements that you are monitoring on a routine basis). Some of those features might reveal some problems or that some alerts need to be produced. To output those to an alerting system, you want to find elements that pass some criteria in the RDD of data that you’re currently looking at. To do that, you are going to iterate over the RDD of results. Beside the alerting, you might also want to publish the state of your application to, for example, feed a data visualization or a dashboard, informing you on more general characteristics of the system that you are currently surveying.

The point of this thought exercise is to envision that computing on an output DStream involves several operations for each and every RDD that composes the final result of your pipeline, despite it being very structured and probably reduced in size from the input data. For that purpose, using the cache to store that final RDD before several iterations occur on it is extremely useful.

When you do several iterations on a cached RDD, while the first cycle takes the same time as the noncached version, while each subsequent iteration takes only a fraction of the time. The reason for that is that although the base data of Spark Streaming is cached in the system, intermediate steps need to be recovered from that base data along the way, using the potentially very long pipeline defined by your application. Retrieving that elaborated data takes time, in every single iteration that is required to process the data as specified by your application, as shown here:

dstream.foreachRDD{ (rdd)  =>
  rdd.cache()
  keys.foreach{ (key) =>
    rdd.filter(elem=> key(elem) == key).saveAsFooBar(...)
  }
  rdd.unpersist()
}

As a consequence, if your DStream or the corresponding RDDs are used multiple times, caching them significantly speeds up the process. However, it is very important to not overtax Spark’s memory management and assume RDDs of a DStream will naturally fall out of cache when your DStream moves to the next RDD, after a batch interval. It is very important that at the end of the iteration over every single RDD of your DStream you think of unpersisting the RDD to let it fall out of cache.

Otherwise, Spark will need to do some relatively clever computation to try to understand which pieces of data it should retain. That particular computation might slow down the results of your application or limit the memory that would be accessible to it.

One last point to consider is that you should not use cache eagerly and everywhere. The cache operation has a cost that might outweigh the benefits if the cached data is not used enough times. In summary, cache is a performance-boosting function that should be used with care.

Speculative Execution

Apache Spark deals with straggler jobs, whether in streaming or batch execution, using speculative execution. This mechanism uses the fact that Spark’s processing puts the same task in the queue of every worker at the same time. As such, it seems reasonable to estimate that workers should require more or less the same amount of time to complete one task. If that is not the case, it’s most often because of one of two reasons:

• Either our dataset is suffering from data skew, in which a few tasks concentrate most of the computation. This is in some cases normal,³ but in most instances a bad situation, that we will want alleviate (e.g., via shuffling our input).

• Or a particular executor is slow because it’s that executor, presenting a case of bad hardware on the node, or if the node is otherwise overloaded in the context of a shared cluster.

If Spark detects this unusually long execution time and has resources available, it has the ability to relaunch on another node the task that is currently running late. This speculative task (which speculates that something has gone wrong with the original) will either finish first and cancel the old job, or be canceled as soon as the former one returns. Overall, this competition between the “tortoise and the hare” yields a better completion time and usage of available resources.

Speculative execution is responsive to four configuration parameters, listed in Table 26-1.