Applied Spark: Designing & Optimizing a Real Job

This is the last chapter. You build one Spark job from a question to a finished answer, the way you would in an interview.

The job is "daily revenue by country". It reads a big orders table and a small customers table.

You do nine steps, in this order.

1. Ask what the report must show.
2. Estimate the size of the data.
3. Design the job on paper.
4. Write the first version.
5. Read the plan.
6. Find the bottleneck in the Spark UI.
7. Fix it and measure again.
8. Make the job safe to rerun.
9. Say the trade-offs.

No new Spark tool appears here. Each step uses an earlier chapter. This table shows where to look when a step feels new.

The data is the shop from earlier chapters. The real rows prove the answer is right. Made-up volumes, always labelled, show how the job behaves at scale.

A good answer starts with questions. A wrong guess here wastes the whole job.

A BI team (business intelligence, the people who build dashboards) asks for "daily revenue by country". This table shows what you ask and the answers we assume.

Now write the answer you expect, before any Spark code. Use the real shop rows and plain SQL.

SELECT o.order_date,
       COALESCE(c.country, 'unknown') AS country,
       SUM(o.amount) AS revenue
FROM orders o
LEFT JOIN customers c ON o.customer_id = c.customer_id
WHERE o.status = 'paid'
GROUP BY o.order_date, COALESCE(c.country, 'unknown')
ORDER BY o.order_date

Size decides every later choice: the join, the partitions and the cost. So you estimate it out loud, with numbers.

Use the made-up cluster from "Spark Configurations": 29 executors with 5 cores each. An executor is a worker process that runs tasks. That is 29 × 5 = 145 cores.

A task is the work on one partition. When all 145 cores run tasks at the same time, we call that one round.

These volumes are made up. Look at the inputs, then at what the code works out.

orders_mb = 1024 * 1024        # 1 TB of orders for one day
tasks = orders_mb // 128       # one task per 128 MB
rounds = -(-tasks // 145)      # round up: tasks / cores
partial_rows = tasks * 200     # at most 200 countries per task
print(tasks, rounds, partial_rows)

Three facts decide the design.

• Orders is huge. You must read less of it and avoid moving it.
• Customers is tiny. About 2 MB is far below the automatic broadcast limit of 10 MB (spark.sql.autoBroadcastJoinThreshold).
• The answer is small: at most 200 rows per day.

Now design the job on paper. Start from where data must move, because moving data over the network is the slow part.

In this job data must move in one place only: the final sum per day and country. This move is called a shuffle. Everything else stays on the machine that read it.

The picture shows the plan we want, from left to right.

Each choice in the picture comes from an earlier chapter. This table lists them.

This is the plan we want. In the next sections we write code, then check if Spark really builds this plan.

The function takes a first day and a last day. A daily run passes the same day for both. A backfill, a run that processes old days again, passes a range.

Here is a first version. It follows the design, but it leaves the join strategy to Spark.

from pyspark.sql import functions as F

def daily_revenue(first_day, last_day):
    orders = (spark.read.parquet("/lake/orders")
        .where(F.col("order_date").between(first_day, last_day))
        .where("status = 'paid'")
        .select("order_date", "customer_id", "amount"))

    customers = (spark.read.parquet("/lake/customers")
        .select("customer_id", "country"))

    return (orders.join(customers, "customer_id", "left")
        .groupBy("order_date",
                 F.coalesce("country", F.lit("unknown")).alias("country"))
        .agg(F.sum("amount").alias("revenue")))

daily_revenue("2026-01-01", "2026-01-31").orderBy("order_date").show()

The logic is right. But right rows on 6 orders do not tell you how the job behaves on 1 TB.

explain() prints the plan and runs no job. Run it before you run the job on 1 TB.

On our cluster, a teammate set spark.sql.autoBroadcastJoinThreshold to -1 to fix another job. That turns automatic broadcast off for everyone.

A plan is read from the bottom up. The bottom lines run first. This is the plan, simplified.

daily_revenue("2026-01-10", "2026-01-10").explain()

AdaptiveSparkPlan isFinalPlan=false
+- HashAggregate [order_date, country], sum(amount)
   +- Exchange hashpartitioning(order_date, country, 200)
      +- HashAggregate [order_date, country], partial_sum(amount)
         +- Project [order_date, coalesce(country, unknown), amount]
            +- SortMergeJoin [customer_id], [customer_id], LeftOuter
               :- Sort [customer_id ASC]
               :  +- Exchange hashpartitioning(customer_id, 200)
               :     +- FileScan parquet orders
               :          PartitionFilters: [order_date >= 2026-01-10, order_date <= 2026-01-10]
               :          PushedFilters: [EqualTo(status,paid)]
               :          ReadSchema: [order_date, customer_id, amount]
               +- Sort [customer_id ASC]
                  +- Exchange hashpartitioning(customer_id, 200)
                     +- FileScan parquet customers

The setting of -1 explains it. Spark never broadcasts by itself on this cluster, so the join uses a shuffle. AQE cannot switch it at run time either, because AQE uses the same limit by default.

Next we run the job and see how slow it is.

A bottleneck is the one part that sets the total time. You find it in the Spark UI.

Run the job for the 1 TB day. Open the Stages tab. All numbers here are made up.

Stage 2 takes 20 of the 38 minutes. Click it. The task summary tells you why.

The tasks are even: the longest is 11 minutes and the median is 9. So this is not skew. "Data Skew & Salting" showed the other shape, one task far longer than the rest.

Spill means memory is full, so Spark writes data to disk and reads it back. The cause here is volume. About 100 GB crosses the network, and 40 GB spills to disk.

The reason is a join with a table of 2 MB. The numbers agree with the plan: 100 GB / 200 tasks is about 512 MB for each task.

The fix is the choice we made in the design: broadcast customers. The broadcast() hint asks Spark for it directly.

The hint works even when the automatic limit is -1. And it keeps working if someone changes that setting again.

from pyspark.sql.functions import broadcast

# before
return (orders.join(customers, "customer_id", "left")

# after
return (orders.join(broadcast(customers), "customer_id", "left")

# the plan now prints
AdaptiveSparkPlan isFinalPlan=false
+- HashAggregate [order_date, country], sum(amount)
   +- Exchange hashpartitioning(order_date, country, 200)
      +- HashAggregate [order_date, country], partial_sum(amount)
         +- Project [order_date, coalesce(country, unknown), amount]
            +- BroadcastHashJoin [customer_id], [customer_id], LeftOuter, BuildRight
               :- FileScan parquet orders
               +- BroadcastExchange
                  +- FileScan parquet customers

Run the job again and open the same tabs. The numbers are made up, as before.

Stage 0 keeps its 8,192 tasks, about 57 rounds, as we estimated. It now takes 12 minutes and does the join inside the same tasks. The final stage takes 1 minute.

Spark reads customers in a small separate job first and copies it to the executors. That job takes seconds, so it is not in the table.

AQE is on by default since Spark 3.2. It merges the small partitions of the final sum, so we change nothing there.

Now stop. The report must be ready by 08:00, and 13 minutes is far inside that. More tuning adds risk and gains nothing.

In production a job often runs more than once. A scheduler retries, or a person reruns one day. "Production Spark: Reliability, Cost & Anti-Patterns" taught the rule: a second run must give the same result.

Our output has one folder per day. Dynamic overwrite replaces only the days in the new data.

Look at a late fix. On 2026-01-20, order 105 was still pending, so the job wrote no rows for that day.

Later, order 105 is paid for 30.00 (made up). You rerun that one day, twice.

spark.conf.set("spark.sql.sources.partitionOverwriteMode", "dynamic")

day = "2026-01-20"
(daily_revenue(day, day)
    .write.mode("overwrite")
    .partitionBy("order_date")
    .parquet("/lake/daily_revenue"))

A second safety net is a total check. The revenue in the output must equal the paid amount in the source.

Look at order 290 (made up). It is paid for 12.00 on 2026-01-22, but customer 290 is not in the customers table yet.

orders = spark.read.parquet("/lake/orders")
src = (orders.where("status = 'paid' AND order_date = '2026-01-22'")
             .agg(F.sum("amount")).first()[0])
out = (daily_revenue("2026-01-22", "2026-01-22")
             .agg(F.sum("revenue")).first()[0])
assert out == src, f"revenue lost: source {src}, output {out}"

Every choice has a price. A strong answer names the price and says when it would change the choice.

This table gives the four main choices of this job.

One more price: the job uses the country today. If Asha moves from IN to DE, a rerun of 2026-01-03 puts her 40.00 under DE. If the BI team wants the country at order time, store it on each order.

AQE stays on. Its price is small: the final plan can differ from explain(), so you read the plan after the run too.

Four points to keep.

• Write the expected answer from real rows before you write Spark code.
• Estimate the size, then list where data must move. Make that part small.
• Read the plan before the run, then the Spark UI after it. Name the cause from numbers.
• Change one thing, measure again, and make the write safe to run twice.

Now rehearse aloud with a timer. Six parts, one minute each, six minutes in total.

1. Minute 1: ask the requirements and say the expected answer.
2. Minute 2: estimate the size with numbers.
3. Minute 3: draw the plan and name the one shuffle.
4. Minute 4: say what you look for in the plan.
5. Minute 5: name the slow stage, its cause and the fix.
6. Minute 6: explain the safe rerun and one trade-off.

Then answer these questions without notes.

1. The plan shows three Exchange lines, and you planned one. Where do the other two come from?
2. Stage 2 has tasks of 9 minutes median and 11 minutes max. Is it skew? Why?
3. Why does the job use a left join, and what does the total check show for order 290 with an inner join?