Spark Testing and Debugging

Spark jobs fail in surprising ways. A small mistake can hide for hours in a large run. Testing on tiny data catches logic errors early, and a handful of debugging habits shorten the search when something still breaks.

The Layman View

A baker tastes a spoonful of batter before baking a hundred cakes. Testing a Spark function on five rows works the same way. A wrong flavour shows up in seconds instead of after a long bake.

Design Code for Testing

Put each transformation in a function that accepts a DataFrame and returns a DataFrame. Functions that read and write files inside the logic resist testing.

from pyspark.sql import functions as F

def add_bonus(df, rate=0.10):
    return df.withColumn("bonus", F.col("salary") * rate)

def keep_active(df):
    return df.filter(F.col("status") == "active")

Read data --> keep_active --> add_bonus --> Write data
              (testable)    (testable)

A Local Test Session

Use pytest with one shared local SparkSession. A small number of shuffle partitions keeps tests fast.

# conftest.py
import pytest
from pyspark.sql import SparkSession

@pytest.fixture(scope="session")
def spark():
    session = (SparkSession.builder.master("local[2]")
               .appName("tests")
               .config("spark.sql.shuffle.partitions", "2")
               .config("spark.ui.enabled", "false")
               .getOrCreate())
    yield session
    session.stop()

Writing a Test

# test_transforms.py
from pyspark.testing import assertDataFrameEqual

def test_add_bonus(spark):
    source   = spark.createDataFrame([("Asha", 1000.0)], ["name", "salary"])
    expected = spark.createDataFrame([("Asha", 1000.0, 100.0)], ["name", "salary", "bonus"])

    assertDataFrameEqual(add_bonus(source), expected)

def test_keep_active(spark):
    source = spark.createDataFrame([("A", "active"), ("B", "closed")], ["id", "status"])
    result = keep_active(source)
    assert result.count() == 1

Include edge cases in the sample rows: null values, empty tables, duplicate keys, and unusual text.

Debugging Tools

• show() prints the first rows so you can inspect values.
• printSchema() reveals wrong types.
• explain() prints the plan Spark prepared.
• count() after each step finds the line where rows vanish.
• sample() shrinks a large table for fast experiments.

df1 = raw.filter(...)          print(df1.count())   # 1,000,000
df2 = df1.join(lookup, "id")   print(df2.count())   # 0   <-- the join lost every row

The count of zero points directly at the join key. Check for type mismatches or extra spaces.

Controlling Log Volume

spark.sparkContext.setLogLevel("WARN")

The levels run from DEBUG through INFO and WARN to ERROR. Use WARN for normal runs and INFO or DEBUG while chasing a problem.

Common Errors and Causes

Error	Usual Cause	Fix
AnalysisException: cannot resolve column	Misspelled or missing column	Check printSchema and spelling
OutOfMemoryError on the driver	collect or toPandas on large data	Use limit, write to storage, or aggregate first
OutOfMemoryError on an executor	Large partitions or skew	Add partitions or fix skew
Task not serializable	Function captures a non-serializable object	Create the object inside the function
Python worker exited unexpectedly	Crash inside a UDF	Test the UDF alone on sample data
FileNotFoundException	Wrong path or missing access	Verify the path from the executors

Reading a Stack Trace

PySpark errors print long Java and Python traces. Scroll to the lines that start with the exception name and read the message. The first "Caused by" line usually names the real problem.

Debugging Routine

• Reproduce the problem on a small sample.
• Print the schema and a few rows after each step.
• Remove steps until the error disappears, then add them back one at a time.
• Check the Spark UI for failed tasks and spill.
• Fix one thing and rerun before you change anything else.