Production Spark: Reliability, Cost & Anti-Patterns

A job that works once in a notebook is not a production job. A production job runs every day, fails safely, and does not waste money.

You already know the tools. This chapter shows how to use them together, as one habit, before a job ships.

Here is where each earlier idea comes back.

The chapter has four parts: safe reruns, what Spark retries, cost, and anti-patterns.

An anti-pattern is a habit that looks fine on small data and fails on big data. Each one here has a symptom, a cause and a fix.

In production, jobs run twice. A scheduler retries a failed run. A person reruns a day by hand.

A write is idempotent when running it twice gives the same result as running it once. "Idempotency, Retries & Exactly-Once" taught this idea.

Look at a monthly job. It writes paid orders into one folder per month. These are the orders.

from pyspark.sql import functions as F

paid = (orders.filter("status = 'paid'")
              .withColumn("month", F.substring("order_date", 1, 7)))

# Not safe: every run adds the rows again
paid.write.mode("append").partitionBy("month").parquet("/lake/paid")

# Safe: a run replaces only the months it contains
spark.conf.set("spark.sql.sources.partitionOverwriteMode", "dynamic")
paid.write.mode("overwrite").partitionBy("month").parquet("/lake/paid")

Dynamic mode matters. The default mode, static, can delete every month in the folder when you overwrite. Dynamic replaces only the months in your new data.

A second way is Delta MERGE, from "Delta Lake & the Lakehouse". You match rows on a key, such as order_id.

MERGE INTO paid_orders AS t
USING new_paid AS s
ON t.order_id = s.order_id
WHEN MATCHED THEN UPDATE SET *
WHEN NOT MATCHED THEN INSERT *

Spark fixes some failures on its own. You must know which ones, because a retry runs your code a second time.

This table shows what fails, what Spark does, and what you must check.

The driver is the one machine that plans the job. It has no spare copy, so it is a single point of failure.

A stream is different. Its checkpoint records how far it got, so a restart continues from there. This is from "Structured Streaming Fundamentals".

Here is the risk inside a task. Spark may run a task twice, so a call to an outside system can happen twice.

def send(rows):
    for r in rows:
        post_to_billing(r.order_id, r.amount)   # outside system

# Risky: if a task retries, its orders are sent again
paid.foreachPartition(send)

# Safer: write to storage, then load into billing by order_id
paid.write.mode("overwrite").parquet("/lake/to_bill")

You pay for machines and for the time they run. So cost has two parts: how many cores you hold, and how long.

Most of the work is already in earlier chapters. This table lists each lever and where you learned it.

Spot (or preemptible) machines are cheap machines the cloud can take back. A lost executor is replaced and its tasks run again, so executors can use them.

The driver cannot use them. If the driver is taken back, the whole job stops.

Idle time is the easiest saving. Take the cluster from "Spark Configurations": 29 executors of 5 cores, which is 145 cores.

The job times in this example are made up.

# A job needs 2 hours of this cluster each day.
cores = 29 * 5                       # 145 cores
always_on = cores * 24               # cluster never stops
auto_stop = cores * 2                # cluster stops when the job ends
print(always_on, auto_stop, always_on // auto_stop)

Now the data side. Say a table has 36 monthly folders (made up). A query that filters on one month reads 1 folder out of 36.

File size matters too. Say a 10 GB table is stored in 10,000 files (made up). Each file is about 1 MB, and each file adds work to open.

As a guideline, aim for files of about 128 MB. Then 10 GB is about 80 files.

These anti-patterns all squeeze big data through one place. The driver is one machine, and one partition is worked on by one core.

The first is collect(). It brings every row to the driver. Look at a sum done in a Python loop, and the fix.

# Anti-pattern: every row travels to the driver
total = 0
for r in orders.collect():
    if r.status == "paid":
        total += r.amount

# Fix: Spark adds the numbers, the driver gets one row
total = (orders.filter("status = 'paid'")
               .agg(F.sum("amount")).first()[0])

The symptom is an out-of-memory error on the driver. Or Spark stops the job when the results are bigger than spark.driver.maxResultSize, which is 1g by default.

toPandas() has the same cause, because it collects into one pandas table. The fix is the same: filter, aggregate or write in Spark. Bring back only a small result.

The second trap is repartition(1). It puts all data in one partition, so one core does all the work.

# Anti-pattern: one task writes everything
paid.repartition(1).write.parquet("/lake/paid_one")

# Fix: pick the count from the size (data size / 128 MB)
paid.repartition(80).write.parquet("/lake/paid_80")

The opposite problem is too many small files. Think of the 10,000 files of about 1 MB from the cost section. Reads are slow because Spark must open each file.

The fix is to write fewer, larger files. Use the same size calculation, or compact the table. Delta has OPTIMIZE for this.

These anti-patterns make Spark do more work than it needs. The first is a Python UDF where a built-in function exists.

A UDF is a function you write yourself. Spark sends every row to a Python process and back, and the optimizer cannot look inside it.

# Anti-pattern: a Python UDF for a simple test
@F.udf("string")
def label(status):
    return "ok" if status == "paid" else "no"

orders.select("order_id", label("status").alias("label"))

# Fix: a built-in expression
orders.select("order_id",
    F.when(F.col("status") == "paid", "ok").otherwise("no").alias("label"))

The second trap is groupByKey on an RDD. It sends every value across the network before it adds anything.

reduceByKey adds the values inside each partition first. Then each partition sends one number per key. A DataFrame groupBy does this for you.

pairs = (orders.filter("status = 'paid'").rdd
               .map(lambda r: (r.customer_id, float(r.amount))))

# Anti-pattern: moves every amount, then adds
pairs.groupByKey().mapValues(sum).collect()

# Fix: adds in each partition first
pairs.reduceByKey(lambda a, b: a + b).collect()

The third trap is a Python loop over keys. It starts one Spark job for each key.

# Anti-pattern: one Spark job per customer
for cid in [1, 2, 3, 4]:
    print(cid, orders.filter(F.col("customer_id") == cid).count())

# Fix: one job for all customers
orders.groupBy("customer_id").count().show()

Three more problems come from settings and tuning. Each one has a symptom you can see in the Spark UI.

from pyspark.sql import functions as F

# 1. Cache everything
orders.cache()                   # used once: it only fills memory

# 2. A hard-coded setting
spark.conf.set("spark.sql.shuffle.partitions", "8")   # tuned on a laptop

# 3. Skew: one key has most of the rows
orders.groupBy("customer_id").count().orderBy(F.desc("count"))   # find the hot key

Add a data check to this list. A job that exits cleanly but writes zero rows is a silent failure.

Count the rows before you write, and stop the job if the count is wrong. "Data Observability & Monitoring" shows what else to track.

# paid is the batch built in "Make reruns safe"
n = paid.filter("month = '2026-01'").count()
if n == 0:
    raise ValueError("No paid orders for 2026-01")
paid.write.mode("overwrite").partitionBy("month").parquet("/lake/paid")

Four points to keep.

• Make writes safe to run twice: overwrite the partition, or MERGE on a key.
• Spark can run a failed task up to 4 times in total, so code inside a task must be safe to repeat.
• Cost is cores times time. Right-size, stop idle clusters, and read less data.
• Keep big data away from the driver and from single partitions.

Answer these out loud, as in an interview.

1. A job appends paid orders and a scheduler retries it. What is the total, and how do you fix it?
2. Why can executors use spot machines but the driver cannot?
3. What does spark.driver.maxResultSize protect, and what is its default?
4. A job needs a 145-core cluster for 2 hours a day. How many core-hours are wasted if the cluster stays on all day?
5. Name three anti-patterns, their symptom in the Spark UI, and their fix.