Apache Spark WTF??? — Show Me The Cache!!! (Part I)

The Case (2023 Version)

I was part of a team tasked with migrating some streaming workflows. To make things easier and more maintainable, we developed a small, metadata-oriented, and fully extensible common framework. We did a great job, and everything was running smoothly… until we noticed that after about nine hours of runtime, all the jobs started crashing at different times due to baffling OutOfMemory errors.

But … why? Both the streaming jobs and the framework were relatively simple and lightweight. Where was the memory leak? Since the error occurred across all the streaming jobs, it had to be something they all had in common. One of those common elements was … the cache. All the jobs consumed messages from the same Kafka topic and parsed those messages. Although each job discarded rows and parsed fields differently, they ultimately functioned in very similar ways.

Parsing the messages was a time-consuming task, and because the jobs wrote to different sinks, we didn’t want to repeat the parsing process for each write operation. So, we decided to cache the DataFrame produced by the parsing. Could the cache be the culprit? The OutOfMemory error was raised in the driver, not in the executors, so it wasn’t due to data being cached on the executors, that much was clear. But still … it had to be somehow related to that supposedly negligible, damned cached DataFrame. So … why were you being so naughty with us, Spark?

DIY (Do It Yourself)

aaaa

Conducting A Thorough Investigation

Fortunately, not only we were able to reproduce the issue on our local environment (God bless the local environment) but also managed to make it leak memory faster. We used the tools Visual VM — to appreciate how the memory slowly but relentlessly was increasing and to generate heapdumps— and Memory Analyzer — to get the heapdumps from the previous tool, delve deeper into the top-consuming objects and get hints about the suspects — .

The left side shows “Visual VM”, and the right side shows “Memory Analyzer”

With these terrific and handy tools and taking the simplest of our streaming processes to perform tests, we quickly noticed that the number of SparkSessions was mounting up with each micro-batch execution. The culprit was indeed the caching of that dataframe. It seems that every time a dataframe is cached, Spark internally creates a new SparkSession, so in order to fix our OutOfMemory, we only had to keep track of every cache method applied to a dataframe and unpersist them all before the micro-batch finishes.

How I Thought Cache In Spark Worked …

When I think about caching, the first thing that comes to mind is a LRU (Least Recently Used) cache. To me, caching is something a totally extra feature which only purpose is to improve performance in our processes. For example, in Java, caching usually is implemented using Soft or Weak References, that is, memory is used for caching as long as is not needed for our application, but it’s released when required. So, the least I expect from a cache mechanism is to throw an error, not less a memory leak.

According to Spark’s own documentation:

Spark automatically monitors cache usage on each node and drops out old data partitions in a least-recently-used (LRU) fashion. If you would like to manually remove an RDD instead of waiting for it to fall out of the cache, use the RDD.unpersist() method.

Well, the optional “you would like to manually” remove from cache mentioned in the Spark documentation was a blatant lie, because in streaming or long-lasting Spark applications semes to be in fact totally mandatory unless we want to start getting OutOfMemory errors. Moreover, in streaming or long-lasting Pyspark applications not even unpersisting cached or persisted dataframes prevent us from leaking memory and eventual crashes.

But as it happens with all good mystery novels … the puzzle started to piece together at the end …

… And How It Really Works

aaa

Oops … I Fixed It Again?

sssss

---

Did you like this article? Continue reading other interesting episodes from the Apache Spark WTF??? series:

• The Deadly Filter: You’ll never see filters the same way.
• Spark (Broad)Cast A Spell On Me: A broadcast join, one big and one small tables are the only ingredients to this magically astonishing recipe … with totally unexpected results!

---

Sick and tired of debugging some error or performance issue? I might help you! Send me your case! I’m genuinely passionate about intriguing errors and performance challenges — those are the things that keep me up at night and also thinking while walking my dog. I’ve been solving interesting cases for the last 15 years. so … don’t hesitate to reach me out at https://www.linkedin.com/in/angel-alvarez-pascua.