Getting Lost in the Woods

If you haven’t had the (dis)pleasure of log-centric distributed system debugging, let me tell you—it is a real treat. When a process runs out of resources, you are lucky if you get a descriptive log message clueing you in to what happened. Many times you may get a message that not only doesn’t describe the resource shortfall but also makes you think something else is the problem, such as “could not bind to address.”

In the EMR-based data platform, large ingestion jobs would periodically fail while trying to retrieve a block of data from one of the cluster nodes. What is a “large” ingestion job? Good question: our team didn’t monitor data volume, so it wasn’t terribly clear where the cliff was that caused this problem.

Typically when a failure occurred, the on-call engineer had to race against the clock to inspect cluster metrics before they disappeared.³ For example, our team suspected that autoscaling events were related to these failures, but without monitoring, the only access to autoscaling events was the EMR console, which retained only the most recent events. If the critical time point occurred too far in the past, it would no longer be present in the console.

In addition, these jobs could run for several hours, only to fail after wasting half a day of compute resources. The only recourse was to relaunch the job with more capacity than the failed job. How much more? This also was not clear, since cluster configuration wasn’t monitored either, so our team didn’t know what had worked in the past. The engineer assigned to on-call that week would double the number of workers in the cluster configuration and hope for the best. Relatedly, because data size and cluster size weren’t monitored, a lot of money was wasted due to over-provisioning for smaller jobs.

Our team was deep in the woods with no map, compass, or GPS; just a vague idea that if we bushwhacked downhill, we would eventually find a road leading back to the trailhead.

In addition to engineering and cloud costs, the lack of reliability led to data downtime, eroding trust in the platform. In this case, the customers were a somewhat captive audience, but in other cases, these kinds of issues can negatively impact revenue. Who wants to pay for a data platform that doesn’t reliably ingest and surface data?

To summarize, inadequate monitoring resulted in this project incurring unnecessary costs over several areas:

• Engineering hours spent on root cause analysis

• Over-provisioned compute resources

• Extra compute resources for rerunning failed jobs

• Low platform adoption

I presume someone thought this process was less costly than monitoring. Given the system’s size and complexity, I doubt the cost of an entire developer was less expensive, particularly when the most senior developers had to stop their work to help debug especially thorny problems.

Navigation to the Rescue

Regrettably, these issues did not prompt changes to put additional monitoring in place. There are a number of metrics that could have been tracked to improve resilience, customer trust, and cost-effectiveness. Many of the specific metrics mentioned here will be described in more detail in the rest of the chapter.

Data Volume

In Chapter 1, the HoD team had access to a year’s worth of survey data, repeated here in Figure 11-2, to get a sense of ingestion job size. You saw how to use this data to develop an estimate of compute resources. Similar to this initial estimate, monitoring data volume over time will help you refine and redesign pipeline resources as needs change.

Figure 11-2. Survey data volume per month

For example, in Figure 11-2 there is very little bird survey activity from December through February. Based on migratory patterns, bird surveys peak from September through October, and are otherwise at a fairly consistent level from late March through early August. Assuming this trend is observed over earlier and subsequent years, this can help you budget and choose pricing plans wisely. Trends and seasonality can also help with capacity planning, as you have a sense of when you need more or fewer compute resources.

Monitoring source data volume can also help you track potential system issues; a sudden, unexpected decrease in volume could point to a bug causing data to be lost. I’ve seen this happen when expired credentials caused input data volumes to plummet. On the other hand, a sudden spike in volume could strain the provisioned compute resources.

Monitoring data volume as it moves through the pipeline is another place to check for problems. For example, let’s say you have a filtering step that you expect will eliminate 20% of the source data. If you observe that the volume of data exiting the filter step is 60% lower than the source data volume, there might be a bug in your filtering logic.

Reading through these last examples, you may be thinking: “But what if there is nothing wrong with the pipeline? Perhaps these data volume changes are due to changes in the source data.” You’d be right. Oftentimes metrics are most informative when considered as part of a group and with knowledge of pipeline operation and data characteristics.

For example, one pipeline I worked on primarily reformatted the source data. In this case, it’s expected that the volume of input data would roughly match the volume of output data. A significant delta between input and output data would have been cause for concern. Because this was such a cut-and-dried scenario, the pipeline included a size check as part of the data validation.

Another pipeline I worked on had a far less deterministic relationship between input and output data volume. In this case, we combined data volume with other metrics, such as number and type of customers active in the system, to deduce whether a change in data volume warranted closer inspection.

Remember that you have other tools for checking for issues in a data pipeline, such as the validation techniques covered in Chapter 6 and the testing techniques covered in Chapter 7. Consider changes in data volume in light of the other signals you get from the pipeline.

Throughput

With data volume you can also monitor throughput, the amount of data processed over a given time period. Monitoring the speed at which a pipeline processes data can give you a heads-up as to performance changes, system stability, and scaling opportunities.

For batch pipelines, throughput can be measured based on job duration. Returning to the scenario I described in “Getting Lost in the Woods”, if our team had monitored throughput, it would have been apparent that the EMR resourcing strategy was approaching a cliff for larger jobs.

Figure 11-3(A) shows an approximation of this performance issue, plotting data volume versus job duration. Prior to the data points in this graph, the relationship of increased data volume to job duration was roughly linear. While I’ve omitted this from the graph for brevity, establishing a baseline is essential to know what is within the expected range.

The solid line with the x’s shows the increase in data volume for successive ingestion jobs. The dotted line with the square markers shows the corresponding increase in job duration. Notice that while the data volume increases linearly, the job duration increases significantly beyond a certain data volume. Figure 11-3(B) shows the computed throughput based on data volume and job duration.

Figure 11-3. Reduced pipeline throughput as data volume increases, viewed as data volume and job duration changes (A) versus the computed throughput (B)

Table 11-1 shows the data for Figure 11-3, annotated with the percentage changes in data volume and throughput. Notice that the changes in data volume from one job to the next are about the same, but the changes in throughput more than double.

Had our team tracked throughput, the correlation between data volume, runtime, and reliability issues would have been visible before jobs began failing en masse. This would have given the team not only time to proactively address the problem, but also the data to go to the product owner and push for engineering time to make the fixes.

In this example, tracking data volume and job runtime would not have been a significant undertaking. At the most basic level, you can record these metrics in a spreadsheet to look at trends over time. One of the teams I worked on looked at throughput metrics on a weekly basis, using a spreadsheet to identify baseline behavior and deviations.

For streaming pipelines, throughput can be measured by looking at the lag between data acquisition at the beginning of the pipeline and data storage at the end of the pipeline,⁴ as you will see in the next section.

Consumer Lag

In streaming pipelines, the degree to which consumers are keeping up with the volume of messages from producers is an important metric for system stability. Lag is defined as the number of messages published versus the number of messages consumed. To keep things clearer, I’ll refer to consumers as “workers” in this section.

Let’s take a look at a few examples of lag to get a sense of how this metric can help identify data pipeline issues. Figure 11-4 illustrates a lag profile for a system that is increasingly unable to keep up with the inbound messages. I’ll talk through some different data pipeline issues that can cause this to occur.

First, though, a bit about lag profiles. The left side of each case a through d in Figure 11-4 represents the ramp-up of published messages from the producer. The saturation point, where the curve is flat, indicates that the workers are consuming about the same number of messages as the producer is creating. The ramp-down occurs when the publisher has finished producing and the workers consume the remaining messages.

I’m defining healthy lag as lag that gets resolved—that is, the consumer is eventually able to clear all the messages provided by the producer before the next set of messages come in. Cases a and b illustrate healthy lag, where lag goes to zero before the next group of messages are published.

Figure 11-4. Examples of healthy and unhealthy lag

Compared to a, the lag in b is higher, indicating that more messages are building up from the publisher before the workers process them. You can also see this from the narrower saturation duration. Another thing to note about b is the slope and duration of the ramp-down period. These are longer than a, another indication that it’s taking longer for all the messages to get consumed.

Unhealthy lag occurs when the consumer can’t keep up and messages start to pile up. Notice that in the c and d cases, the lag does not return to zero. If this trend continues, there will be a significant backlog of messages. In c and d, the trend of higher lag over a longer time period is worsening relative to cases a and b.

Why does a message backlog matter? For one thing, it means data isn’t getting processed as quickly, which can impact performance and SLAs. This can be especially detrimental when sub-second latency is required. From a cost-effectiveness point of view, very low-latency systems are expensive and in many cases are not necessary. To find the right trade-off between cost and latency, work with customers and product owners to determine a reasonable SLA. You can then monitor latency and tune resources to hit the target. If it’s going to take a lot of extra resources (read: costs) to hit an SLA, you can bring that information back to stakeholders to reevaluate.

A message backlog consumes resources on the message broker, which is responsible for handling the message exchange between publishers and workers. What starts as sluggish performance due to message backlog can tip over into exhausting broker resources.

You can preempt this failure mechanism by alerting when lag gets into concerning territory. By monitoring over the long term, you’ll get to know how much lag the system can clear and what level of lag indicates you might be headed for trouble. Recalling alert fatigue, be sure to consider whether alert thresholds could be tripped by noncritical events, such as heavyweight testing.

Why would the lag changes in Figure 11-4 occur in a data pipeline? It could be that more messages are getting produced as a result of higher data volume. In this case, you would see more messages from the publisher and may need to add more workers to reduce the lag. Thinking back to scaling in Chapter 2, you could address this with scaling rules based on message volume and backlog. Scale out to clear pending messages, and scale back in when the backlog is clear.

On the other hand, changes in data volume don’t necessarily mean you want to ingest all the data. I was working on a data pipeline where one of the data sources suddenly started producing significantly more data, resulting in the pipeline lag increasing dramatically. This prompted a developer to take a look at the source data, ultimately determining that the higher volume was due to new data features that weren’t relevant to the product. In this case, the extra data was filtered at the source, returning the data volume to its previous size and resolving the unhealthy lag.

Another possibility is that the worker has slowed down. I’ve seen this happen as a secondary effect of increased data volumes, where data transformation time increases as a result of processing more data. Lag can also build up if the worker is waiting on an external resource. Perhaps another service your pipeline interacts with is under-provisioned or starts throttling back connections. It’s essential to root out these bottlenecks in real-time systems, as they limit your operational and scaling capacity.

I’ve seen this scenario in the field, when working on a pipeline that interacted with several different databases. Our team observed that pipeline jobs started failing suddenly, as if the system was at a tipping point and had just gone over the edge.

Some system monitoring was in place at the time, so our team could see that job-processing times had increased considerably. Unfortunately, there wasn’t enough granularity in our monitoring to determine where exactly the slowdown was coming from. Investigating resource utilization on the databases didn’t turn up anything. Much like a game of Whac-A-Mole, the path forward involved progressively disabling different parts of the pipeline to figure out where the issue was.

The root cause was resource contention at one of the databases. The queries executed by the pipeline were taking a very long time as a result of some recent data changes. The issue wasn’t that the databases didn’t have enough resources to run the queries; rather, the number of workers and concurrency settings per worker needed to be increased to take advantage of the available database resources. In other words, the current worker settings were insufficient now that database queries were taking longer.

This story is a good example of the multidimensional seesaw of managing data pipeline operation. The system was performing well until one of the many data pipeline dependencies changed, throwing the balance off. While I mentioned increasing worker count and concurrency as a solution, our team did more than that. The data changes that precipitated the event were addressed by rethinking our data storage strategy to minimize database contention. The capacity planning for the database was revisited; as with more consumer throughput, the stress point would move from the workers to the database. These changes spanned infrastructure, data modeling, and code.

So far, I’ve focused on unhealthy lag, what it can mean in data pipelines, and strategies to mitigate it. It’s also possible for lag to show where you are over-provisioned, that is, where you’ve allocated more consumer resources than necessary.

Figure 11-5. Minimal lag example

This can be tricky territory, as you saw in the example of database contention that lag is only part of the story when it comes to data pipeline performance. Notice in Figure 11-5 that the lag is very small compared to the amount of time between message publishing events. If you consistently see very little lag compared to the amount of time available for processing, you may be able to reduce worker resources. To make sure you don’t prematurely scale down and create performance issues, keep the following in mind:

Broker resources

Fewer workers means more messages are sitting on the broker.

Consumer variability

Are there scenarios where the consumer could take longer to process a message that could lead to unhealthy lag?

Producer consistency

Are the received messages consistent, such as from a well-characterized, consistent data source, or could messages change without notice?

Alerting

Choose a threshold to alert on unhealthy lag to give you time to reevaluate resourcing.

As an example, I was working on a pipeline where our team observed minimal lag and tried downsizing the Spark deployment to see whether costs could be saved. With the reduced deployment, the lag increased to near unhealthy levels, so the downsizing wasn’t possible. While our team couldn’t cut costs in this situation, this example illustrates how to look for signals in lag and assess whether changes can be made.

Understanding the Bounds

You may have heard processes referred to as being “CPU bound” or “memory bound” to communicate what is limiting performance. For example, the performance of joining large datasets in memory is likely to be bound by the memory available in the container or cluster where the process is running.

Knowing which attribute is limiting performance will show you which resource changes will help and which will not. In the large join example, throwing more CPU capacity at the problem is unlikely to help, unless you’re trying to help your CSP make more money.

In some cases, you can exacerbate issues by over-provisioning the wrong resource. Let’s say you have a data-processing stage that acquires data from another resource. If this process is I/O bound, meaning its performance is limited by waiting for the data source, adding more CPU could make the problem worse. Consider that the faster the data processing stage runs, the more requests it can make to the data source. If the data source is responding slowly due to its own resource limitations, the delay between request and response could increase.

Understanding how pipelines process data can help you determine what is bounding your process. In the EMR pipeline example at the beginning of this chapter, the source data was copied to disk and then processed in memory with Spark. Given this, disk and memory were logical assumptions for what bounded that process. You also saw an example in Chapter 1 of how choosing a high-memory machine type performed more quickly and more cheaply for another Spark workload, hinting at the memory-bound nature of that process.⁵

Fortunately, resource monitoring is pretty well established. It’s likely you can access this information within your infrastructure. The exception to this is managed pipeline services, where you might be able to get insight into system-level metrics such as lag, but you may not have access to more in-depth metrics such as resource utilization.

Understanding Reliability Impacts

In the most severe cases, processes that overrun resource allocations can keel over. This can impact a pipeline directly if a stage runs out of resources or indirectly if a service the pipeline communicates with is killed due to out-of-resource issues. In one case in which I saw this happen with a database, the pipeline began failing as a result of the database running out of resources and being unable to respond to requests. This is where idempotency and retries come into play, as you learned about in Chapter 4.

If you’re working in a containerized environment, monitoring service restarts can help identify inadequate resourcing. A container that consistently restarts unexpectedly may be running into resource issues.

A process getting killed repeatedly is a pretty direct sign that something is wrong. Sometimes resource shortfalls can be more difficult to detect. In one situation, I observed a Spark job get stuck in a cycle of stalling, then making progress when garbage collection ran. The garbage collection freed up just enough memory for the job to move forward a little before running out of resources again. Sometimes the job managed to succeed after several hours, but other times it failed.

Memory leaks are another source of instability that isn’t readily apparent. Memory utilization that slowly but steadily increases over time can be a sign of a leak. This is a situation where a longer observation window is helpful; if you don’t monitor memory long enough to see the leak, you won’t be aware that it’s happening.

On the other end of the observation spectrum is how frequently metrics are evaluated. This is a trade-off with cost, as frequent evaluation is expensive, and with usability, as temporary resource spikes may go undetected because the evaluation window is too long.

Similar to the Airflow example in “Worker Utilization”, addressing resource shortfalls by adding more resources isn’t necessarily the most cost-effective approach. I worked on a system that used Redis to cache results for common queries. At some point, queries started to fail. Looking into the issue, it turned out that Redis was running out of memory. By inspecting memory use over time, our Ops team noticed that Redis was holding on to a lot of memory outside of the time frame where the queries ran.

This finding prompted the developers to get involved to figure out why this was happening. The developers reevaluated the time-to-live (TTL), a duration after which a Redis entry is deleted. It turned out that the TTL was too long, so one optimization was reducing this to a more reasonable duration.

The investigation also revealed that a handful of Redis entries were using up most of the memory. After further discussion, it was determined that these entries were no longer needed, having been part of a deprecated feature. With the TTL adjustments and unused entries removed, Redis memory was reduced, which not only fixed the query failures but also reduced compute costs. Had our team simply provisioned more resources to fix the memory issue, we would have been wasting money.

Pipeline Stage Duration

To dig into pipeline performance at a lower level, let’s revisit the HoD batch pipeline from previous chapters, repeated here in Figure 11-9.

Figure 11-9. HoD batch pipeline

Let’s say you’ve been monitoring the overall runtime of the pipeline for a while and recently noticed some deviations from the expected baseline. Table 11-2 shows the baseline runtime and two cases that have started to appear frequently. In Case 1 the runtime is about three times the expected runtime, and in Case 2 it is about 30 times the expected runtime.

The runtime on its own tells you something is amiss, but without more detail it’s difficult to know what is going on. Before reading on, think about some ways you would try to get to the bottom of these longer runtimes.

This is the situation I was presented with in the database contention problem I shared in “Consumer Lag”. Something was taking an unusually long time, but it wasn’t clear what it was.

Table 11-3 shows some examples of the more in-depth picture you get when looking at runtimes within the pipeline, with the pipeline stage names abbreviated.

Table 11-3 makes it clear where the slowdowns are, but typically you would first observe this information in a dashboard, visually inspecting metrics to look for anomalies. Figure 11-10 shows an idealized trace for the Baseline and Case 1. The top graph shows the overall throughput of the pipeline, and the bottom graph breaks down traces by pipeline stage. For readability, the stage traces are offset on the y>-axis, and I’ve limited the stage traces to zip code (solid line), validate (dashed line), and enrich (dotted line).

Figure 11-10. Baseline versus Case 1: dashboard view

Getting familiar with the typical characteristics of metrics will help you debug at a glance whether something isn’t as expected. In the overall view at the top of Figure 11-10, it’s clear that Case 1 has very different throughput than Baseline does, which would prompt you to take a look at the stage traces for further inspection. At the stage level, you can see that the throughput of the Enrich stage is lower than Baseline, and as a result, it takes longer for this stage to complete.

Returning to Table 11-3, let’s think through the different cases. Interestingly, Case 2 appears to have two different underlying causes. At this level, you can start drawing on your knowledge of pipeline operation to consider what might cause these deltas. For example, perhaps you recall that there is a five-minute retry when connection issues occur with the zip code API or the database.

A further step you can take here is to consult the logs, as you saw in Chapter 10. If you’ve left yourself some good breadcrumbs, you might see connectivity errors and retry attempts corresponding to the Case 2 scenarios. From there, you can go back to the API and database to troubleshoot further.

What about Case 1? If there are no connectivity errors, what else could be the problem? If you’ve logged or created metrics for data volume at each pipeline stage, as suggested in “Data Volume”, this would be another place to look. Another option for sub-stage level observation is profiling.

Profiling

If you need more fine-grained visibility into where a pipeline is spending time, profiling can help you get to a much lower level of detail. A word of caution: profiling can suffer from the observer effect, where the act of measuring runtime can impact the overall duration. Whereas job- and stage-level runtime observability can leverage metadata, such as the task runtime metrics made available by systems like Airflow and Celery, profiling depends on putting probes within the code itself.

Profiling can be expensive from a cost and performance perspective. If you choose to implement profiling, be able to turn it on selectively—for example, on a per-job or per-customer basis, or perhaps only in a test environment where performance isn’t critical.

How profiling is implemented and the level of detail required impact cost. If your interests are limited to method runtime, using a wall timer and logging the metric is an inexpensive option. I’ve seen this approach used to generate JSON logs that were then analyzed in Google BigQuery, similar to this example where the event_id was traceable throughout the pipeline (times here are in seconds):

{
    "event_id": "bird-23ba"
    "total_time":10,
    "enrich_with_social": 7,
    "store_result": 0.4,
    ...
}

If you want to monitor runtime continuously, setting up a custom time-series metric is another option. Each function you monitor will be a new trace, which can multiply by other factors depending on how the metrics are labeled. More about that shortly.

Many observability services offer application performance monitoring (APM), which includes the ability to profile the full range of resource metrics down to the class and method levels. This can be quite expensive, but it can also be very illuminating to help tune performance.

If you’ve implemented the code design strategies described in Chapter 6, you’ll already have a codebase set up to easily probe the amount of time it takes for discrete processes to run. Continuing Case 1 from Table 11-3, assuming you’ve separated the concerns of interacting with the database from the join performed in “Enrich with social,” you can probe these processes separately, as shown in Table 11-4.

With this level of detail, you can see that, while both the merge and store processes have increased compared to the baseline, the merge has increased by 20 times compared to two times for storing the result.

Errors to Watch Out For

So far, you’ve seen ways to monitor data pipelines to help you keep ahead of performance and reliability issues, while looking for opportunities to reduce costs. Inevitably, you will also have to deal with failures and errors, the reporting of which is an important aspect of monitoring.

In some sense, you can think about error monitoring as going the last mile when it comes to building data pipelines. A lot of times, so much emphasis is on design and development that it’s easy to forget that making a connection from the pipeline to monitoring tools is essential to deal with failures quickly and effectively.

Depending on your infrastructure, you may already have access to some error metrics. For example, Airflow provides various failure metrics, as does Spark. While it’s helpful to know that an Airflow DAG failed, being able to convey why it failed can expedite prioritization and debug. In addition, frequent failures of a particular type can be signs that the design needs to be updated.