Apache Spark WTF??? — 💪 SQL Goes Hardcore (Part II: The Drills) 🥵

“Si captas todas estas palabras que vas a leer, 
 podrás decir que leíste a un buen data engineer”

“If you catch all these words you are about to read,
 you can say that you read a good data engineer.”
 (adapted from “El bum del jarkore” by VKR)

---

👋 Welcome to the Bootcamp

Soldier, it’s time to get your hands dirty. Before we tackle the drills, we need our ammunition: synthetic datasets ready for battle.

We’ll use tpchgen-cli to generate the tables in Parquet format:

tpchgen-cli -s 30 --tables lineitem --format parquet --output-dir data

This will produce 30GB of TPC-H data, our proving ground for every drill ahead. Each query we run, every optimization we test, will be forged on this dataset—no shortcuts, no excuses.

We’ll go section by section, tackling each drill and its corresponding pitfalls from the previous article. And who knows? Along the way, maybe we’ll pick up a new trick or two to sharpen our Spark skills.

---

📦 Drill #1: Constant Folding

Review

In Spark, “foldable” refers to an expression that can be computed at compile-time (i.e., during query planning) rather than at runtime for every row.

An expression is foldable if Spark can evaluate it once on the driver while building the query plan. Spark can then replace the expression with its literal value in the physical plan.

########################  
# Foldable Expressions #  
########################  
F.lit(5) + F.lit(10)  # Spark can compute 15 at plan time  
F.lit("hello") || F.lit(" world")  # "hello world" is known at plan time  
  
############################  
# Non-foldable Expressions #  
############################  
F.rand()            # Random every row ➜ cannot fold  
F.current_timestamp()  # Changes every row ➜ cannot fold  
F.col("L_EXTENDEDPRICE") + 10  # Depends on column ➜ cannot fold

Why It Matters:

• Performance: Foldable expressions are computed once, saving CPU cycles.
• Consistency: Foldable expressions produce the same value across all rows.
• Optimization: Spark’s Catalyst optimizer can simplify or push down foldable expressions, e.g., in filters or projections.

PITFALL 1 — Non-Deterministic Functions

def pitfall_nondeterministic():  
    # Pitfall: rand() inside an expression ➜ recomputed for every row  
    df = lineitem.withColumn(  
        "random_multiplier",  
        F.col("L_EXTENDEDPRICE") * F.rand()  
    )  
  
    utils.write_noop(df)

Using rand() directly inside a column expression is risky because Spark recomputes a new random value for every row. This not only makes our results inconsistent but can also slow down query execution.

When we compute the random value once on the driver and broadcast it to all rows using F.lit(), Spark treats it as a true constant, which it can fold and optimize efficiently.

Takeaway: Spark trusts literal values more than expressions that vary per row, so whenever possible, compute constants up front rather than relying on row-wise calculations.

PITFALL 2 — Over-Aggressive Folding / Dangerous Casts

Over-aggressive constant folding happens when Spark tries to simplify expressions at query planning time but does so in a way that breaks correctness or causes performance problems. Usually, this happens if Spark assumes some value is safe to fold, but it’s actually not — or if you include “unsafe” literals in expressions.

def pitfall_dangerous_cast():  
    # Pitfall: inserting 'not-a-number' in a branch of a CASE expression  
    # may cause planning failures when Spark attempts constant folding  
    df = lineitem.withColumn(  
        "safe_cast",  
        F.when(F.col("L_EXTENDEDPRICE") > 0, F.lit("not-a-number"))  
         .otherwise(F.lit("0"))  
    )  
    df = df.withColumn(  
        "cast_test",  
        F.expr("try_cast(safe_cast AS int)")  
    )  
  
    utils.write_noop(df)

What happens:

1. Spark sees "not-a-number" and "0" in the when() expression.
2. During constant folding, Spark tries to precompute or optimize these values.
3. "not-a-number" is not castable to int, so Spark can’t fold this correctly.
4. The query slows dramatically because Spark has to fall back to runtime evaluation for every row. On top of that, Spark internally generates and throws a java.lang.NumberFormatException for each row, which adds significant overhead and further impacts performance.

The safe fix is to replace "not-a-number" with a safe, deterministic literal, ensuring Spark can fold the value correctly and optimize the query.

Takeaway: Over-aggressive folding isn’t inherently bad — it’s Spark trying to be smart — but if we give it unsafe or invalid constants, it can backfire, slowing queries or producing unexpected results. Always feed Spark safe, foldable values.

PITFALL 3 — Constants Aren’t Always Constant

def pitfall_constants():  
    # Pitfall: expression depends on non-foldable pseudo-constants  
    # ➜ each part may be re-evaluated or optimized unexpectedly  
    expr_value = 1 + F.rand() + (F.current_timestamp().cast("double") % 0.01)  
    df = lineitem.withColumn("time_based_discount", expr_value)  
  
    # Compute min/max of the generated column  
    result = df.agg(  
        F.min("time_based_discount").alias("min_time_based_discount"),  
        F.max("time_based_discount").alias("max_time_based_discount")  
    )  
    result.show()

Expressions that combine rand(),current_timestamp() or similar may look like constants, but Spark cannot fold them—they are recomputed for each row. While the performance impact might be minor, the results can be inconsistent across executions.

The fix is to compute the constant once on the driver and apply it to the column using F.lit().

Takeaway: whenever we need predictability, broadcast true constants so Spark can optimize efficiently and our min/max metrics remain consistent.

---

⬇️ Drill #2: Predicate Pushdown

Review

---

✂️ Drill #3: Projection Pruning

Review

---

🔗 Drill #4: Join Reordering

Review

---

➕ Drill #5: Filter Combining

Review

---

🏗️ Drill #6: Subquery Collapse

Review

---

💣 Drill #7: UDFs

Review

---

🎖️ Drill #8: The AQE Wildcard

Review

---

🗂️ Drill #9: PartitionEvaluator

  val USE_PARTITION_EVALUATOR = buildConf("spark.sql.execution.usePartitionEvaluator")  
    .internal()  
    .doc("When true, use PartitionEvaluator to execute SQL operators.")  
    .version("3.5.0")  
    .booleanConf  
    .createWithDefault(false)

https://chatgpt.com/c/695f3ab5-6104-832b-8dba-0805f29e169a

---

🐍🆚☕ Drill #10: Python-side vs Java-side Execution

https://chatgpt.com/c/695f2b0f-55e0-8329-afc2-c904cf727dbf

---

🫡 Dismissed!

---

“¡Kreyentes! ¿Qué? Decidles qué hora es. ¡Es la hora de Spark!”

“Believers! What? Tell them what time it is. It’s Spark time!”
 (adapted from “El bum del jarkore” by VKR)