Memory Management & Tungsten

Spark keeps data in memory while it works. When memory is too small, a job gets slow or it crashes. In this chapter you learn where an executor puts its memory.

You know the executor from "Spark Architecture: Driver, Executors & Cluster Managers". You know caching from "Caching, Persistence & Checkpointing". The chapter "Performance Tuning & Debugging with the Spark UI" covers the Spark UI in full.

You will learn five things.

1. How an executor splits its memory into parts, with real numbers.
2. How running work and cached data share one pool.
3. What memory a single task gets, and what spill is.
4. Why a job fails with out-of-memory, and which fix to try first.
5. What Tungsten is: binary rows and generated code.

We use "Spark 3" defaults. Check the settings on your own version.

An executor is one Spark process on a worker machine. It runs on the JVM, the Java engine.

You set its heap size with spark.executor.memory. The heap is the memory area the JVM manages for objects.

Spark splits the heap into three parts. Reserved memory is 300 MB that Spark keeps for itself.

The unified pool is 60% of the rest. User memory is the remaining 40%.

The unified pool holds two kinds of data. Execution memory holds data while a shuffle, join, sort or aggregation runs. Storage memory holds cached data.

Take an executor with 8g. Look at the settings, then the numbers.

heap_mb = 8 * 1024                  # spark.executor.memory = 8g -> 8192
reserved_mb = 300                   # fixed
usable_mb = heap_mb - reserved_mb   # 7892
pool_mb = usable_mb * 0.6           # spark.memory.fraction = 0.6
storage_line_mb = pool_mb * 0.5     # spark.memory.storageFraction = 0.5
user_mb = usable_mb * 0.4           # the rest of the usable heap
overhead_mb = max(384, heap_mb * 0.10)   # spark.executor.memoryOverhead

print(usable_mb, round(pool_mb), round(storage_line_mb), round(user_mb), round(overhead_mb))
# 7892 4735 2368 3157 819

The pool is 4735 MB. The "protected" 2368 MB is the part of the pool that cached data can keep. The next section explains it.

User memory is for your own objects in JVM code, for example in Scala RDD functions. Spark does not manage it. If you fill it, the executor can run out of memory.

Overhead is the 819 MB outside the heap. The section after next explains it.

The two kinds of data do not have fixed sizes. They share the 4735 MB pool and borrow free space from each other.

There are two rules. Execution can push cached blocks out of memory, but only down to the protected line. Storage can never push running work out.

Follow one executor over three moments. The pool is 4735 MB. The protected line is 2368 MB.

At moment 2, storage uses 632 MB above the line. At moment 3, only that 632 MB can be taken.

The sort gets 1000 + 735 free + 632 = 2367 MB. The rest of the sort goes to disk.

A dropped block is not lost. With MEMORY_AND_DISK it goes to disk.

With MEMORY_ONLY Spark builds it again when needed. You saw both in the caching chapter.

An executor needs memory that is not in the JVM heap. Threads, network buffers and Python workers use it. Spark asks for it with spark.executor.memoryOverhead.

A container is the block of memory and CPU that YARN or Kubernetes gives one executor.

The default is the larger of 384 MB and 10% of spark.executor.memory. For the 8g executor that is 819 MB, so the container needs 8192 + 819 = 9011 MB. On Kubernetes, a PySpark job uses 40% by default, not 10%.

A Python worker is a separate Python process on the same machine. PySpark starts one to run your Python functions in a query. It lives outside the heap, so it uses overhead memory.

By default Spark does not limit it. You can set spark.executor.pyspark.memory to give Python its own limit.

You can also move some Spark data off the heap. Set spark.memory.offHeap.enabled to true and spark.memory.offHeap.size to a size, for example 2g. Both are off by default.

Spark then keeps execution data in that extra area. Cached data goes there only with the OFF_HEAP storage level.

The JVM does not clean up this area, so its clean-up pauses can be shorter. The container must still have room for it.

An executor runs many tasks at the same time, one for each core. All of them use the same execution memory.

Say N tasks are running now. Each task gets at least 1/(2N) and at most 1/N of the execution memory. N counts the tasks that run now, not the cores.

Take the 8g executor with the whole 4735 MB free for execution. Read the code, then the table.

pool_mb = 4735                  # the pool of the 8g executor
for n in [1, 2, 4, 8]:          # n = tasks running now
    print(n, round(pool_mb / (2 * n)), round(pool_mb / n))
# 1 2368 4735
# 2 1184 2368
# 4 592 1184
# 8 296 592

Now test partitions of different sizes on this executor. A partition is the piece of data one task reads. Use the 4-task row, 592 to 1184 MB.

These sizes are rough. A sort or a join needs different amounts, and data in memory is often bigger than the file on disk. But the idea holds: a big partition on a busy executor does not fit.

Spill means a task writes data to disk because it does not fit in memory. The job goes on, but slower.

Out-of-memory, or OOM, means a task needs memory that it cannot get. The task fails.

You see spill in the Spark UI. Open a stage and look at the task table. Two columns show it.

• Spill (Memory): the size of the spilled data while it was in memory.
• Spill (Disk): the size of the same data on disk. It is often smaller, because it is compressed.

Here is a made-up stage. Orders history is shuffled for a sort, and every task spills.

# Spark UI, stage 3, four of its 200 tasks (the rest look the same)
# Task  Duration  Spill (Memory)  Spill (Disk)
#  16      3.1 min     1.9 GiB        400 MiB
#  17      3.2 min     2.0 GiB        410 MiB
#  18      3.4 min     2.1 GiB        430 MiB
#  19      3.0 min     1.8 GiB        380 MiB

# The fix: more, smaller partitions
spark.conf.set("spark.sql.shuffle.partitions", 800)   # was 200

More partitions made each piece smaller, so each task fits in memory. You do not need more memory.

Adaptive Query Execution, or AQE, changes the plan while the query runs. With AQE you can start with many partitions, and Spark merges the small ones. See "Adaptive Query Execution & Cost-Based Optimization".

Spill and OOM have different causes. This table shows the common ones and the first fix.

Tungsten is a set of changes in the Spark engine. Its goal is to use memory and the CPU better.

The Databricks team started it for Spark. It works mainly for DataFrames and Spark SQL. RDD code gets little of it.

Tungsten has three ideas.

1. Binary rows: Spark stores a row as compact bytes, not as Java objects.
2. Fast sort: Spark sorts a short key stored next to each row pointer, so the CPU reads less memory.
3. Generated code: Spark writes one Java function for a group of steps. The next section shows it.

Start with binary rows. Java objects carry extra data for each object, and the JVM has to clean them up. This cleaning is garbage collection, or GC, and it pauses work.

Spark stores a row in a format called UnsafeRow. It is a block of bytes with a fixed slot for each column. Take order 101.

# columns: order_id, customer_id, status, amount, order_date   (amount is decimal(10,2))
# row:     101       1            'paid'  40.00   2026-01-03
#
# 8 bytes   null bits (one bit for each column)
# 8 bytes   slot for order_id
# 8 bytes   slot for customer_id
# 8 bytes   slot for status (it points to the text)
# 8 bytes   slot for amount (a decimal(10,2) fits in 8 bytes)
# 8 bytes   slot for order_date
# 8 bytes   the text "paid", padded to 8 bytes
# total = 8 + 5 x 8 + 8 = 56 bytes

Spark reads and compares these bytes directly. It makes fewer objects, so GC has less to do.

By default the bytes live in the JVM heap. Off-heap memory, from the section "Memory outside the heap", is an option.

The Spark project and Databricks describe the aim as less memory and less GC. This chapter gives no speed numbers. Measure on your own data.

Think of a stage that filters rows and then sums them. A simple engine calls a function for each step, for each row. That is many calls.

Spark instead writes one Java function for the whole group of steps. This is whole-stage code generation, or codegen. It runs one loop over the rows.

The idea looks like this. This is not the real generated code.

sums = {}
for row in scan_orders():                       # read a row
    if row.status == 'paid':                    # filter
        sums[row.customer_id] = sums.get(row.customer_id, 0) + row.amount   # partial sum

You can see which steps were fused. Run the query once, then call explain(). Look for a star and a number, like *(1).

Look at the paid revenue by customer. This is a simplified plan.

df = orders.filter("status = 'paid'").groupBy("customer_id").sum("amount")
df.collect()   # run it first: with AQE on, the stars show only after a run
df.explain()

*(2) HashAggregate(keys=[customer_id], functions=[sum(amount)])
+- Exchange hashpartitioning(customer_id, 200)
   +- *(1) HashAggregate(keys=[customer_id], functions=[partial_sum(amount)])
      +- *(1) Filter (isnotnull(status) AND (status = paid))
         +- *(1) ColumnarToRow
            +- FileScan parquet orders

The Exchange line has no star. A shuffle cannot be fused, because it moves rows between machines. So a shuffle ends one block and starts the next.

To read the generated Java, call df.explain("codegen"). Do this only when you debug.

Some plans have no stars. A star is missing when a step cannot be generated. Spark also skips codegen for a very wide table, over 100 columns by default.

Since Spark 3.2, Adaptive Query Execution is on by default. If the query has a shuffle, explain() then shows no stars until the query has run.

RDD code has no stars. An RDD runs your functions on plain objects, Python objects in PySpark. Catalyst, the Spark query planner, does not plan it.

This is one reason DataFrames are faster. The next section shows a step that breaks the fused code.

A UDF, or user-defined function, is a function you write and use in a query. A Python UDF runs in a Python worker, outside the JVM. So Spark must send every row there and get the answer back.

Spark has to turn the row into bytes, send it, run your function, and send the result back. This work is called serialization cost. It is paid for every row.

Take a label for each order: "big" if amount is 50 or more, else "small". Here is the same label as a Python UDF and as built-in functions.

from pyspark.sql.functions import udf, when, col

@udf("string")
def size_udf(amount):
    return "big" if amount is not None and amount >= 50 else "small"

with_udf = orders.select("order_id", size_udf("amount").alias("size"))
built_in = orders.select("order_id",
                         when(col("amount") >= 50, "big").otherwise("small").alias("size"))

with_udf.explain()
# *(2) Project [order_id, pythonUDF0 AS size]
# +- BatchEvalPython [size_udf(amount)], [pythonUDF0]
#    +- *(1) Project [order_id, amount]
#       +- *(1) ColumnarToRow
#          +- FileScan parquet orders

The built-in version stays inside one generated function. The UDF version sends rows to Python. It also uses Python worker memory, which is outside the heap.

A pandas UDF is cheaper. It sends rows in batches that use the Arrow format.

The default batch is up to 10000 rows, set by spark.sql.execution.arrow.maxRecordsPerBatch. The plan then shows ArrowEvalPython.

The best choice is a built-in function when one exists. The chapter "UDFs & the Pandas API on Spark" covers UDFs in full.

Start from what you see. Try the fixes in the order of the table, from cheap to costly.

Raise memory last. More memory costs money on every run, and it may not fix the cause.

Four points to keep.

• The heap has reserved memory, the unified pool (60%) and user memory. Overhead sits outside the heap.
• Execution can push cached blocks out, only down to the protected line. Storage cannot push out running work.
• More tasks at once means less memory for each one. Smaller partitions fix most spill.
• Tungsten stores rows as bytes and generates code for a stage. Python UDFs break that chain.

Answer these without looking back.

1. An executor has 4g, which is 4096 MB. How big is the unified pool? Hint: subtract 300 first.
2. Why can a sort spill when a lot of the pool holds cached data?
3. What is the difference between Spill (Memory) and Spill (Disk)?
4. What does the star in *(1) mean, and why has an Exchange no star?