Module 1 · Foundations & Architecture

How PySpark works inside

Beginner 16 min read Explains 90% of PySpark perf

Here is the fact that surprises everyone: Spark is not written in Python, and your Python code does not process your data. Spark is a Scala/Java system. PySpark is a thin client that drives it. Once you understand exactly where the boundary sits, you will know — without measuring — which of your operations are fast and which are 10× slower than they need to be.

After this lesson you can…

  • Explain the Py4J bridge and what crosses it (hint: not your data).
  • Say why df.filter(col("x") > 5) is as fast in Python as in Scala, but a Python UDF is not.
  • Describe what a Python worker process is and when one gets started.
  • Read a Py4JJavaError traceback without panicking.
  • Explain what Apache Arrow does for PySpark and how to enable it.

1. The mental picture

🗣️

The interpreter at a negotiation

You are negotiating with a factory abroad. You speak Python; the factory floor speaks JVM. An interpreter (Py4J) sits next to you. You say "increase order to 500 units" and the interpreter relays it — a short sentence crosses the language barrier, then the whole factory works in its own language at full speed. That is a DataFrame operation.

But if you insist on personally inspecting each of the 500 units, every unit must be carried across the room to you, described, and carried back. The factory sits idle while you work. That is a Python UDF. The interpreter is not the problem — how much you push across the barrier is the problem.

2. The driver side: Py4J

When you call SparkSession.builder.getOrCreate(), PySpark launches a JVM as a subprocess of your Python process and opens a local socket to it. That socket is Py4J, a small library that lets Python call Java methods and hold references to Java objects.

So when you write df.filter(...), your Python DataFrame object is a very thin wrapper holding a handle to a real Java Dataset object living in the JVM. The call is forwarded, the JVM builds the plan, and Python gets back a new handle. Your data never enters the Python process on the driver.

DRIVER MACHINE — two processes, one socket Python process (CPython) your_job.py df = spark.read.parquet(...) df is a wrapper holding a JVM handle JVM process (the real Spark) org.apache.spark.sql.Dataset@3f2a SparkContext · Catalyst · schedulers holds the plan and all the metadata "call filter(...) on obj 3f2a" "here is handle 3f2b" Py4J socket ✓ What crosses the bridge · Method names and small arguments · Object handles (pointers, not contents) · Config values and schema definitions · The small result of show() / collect() Tiny. Happens once per API call. Free. ✗ What does NOT cross · Your rows · Your partitions · Any intermediate result · Anything at all during execution …unless you write a Python UDF. See section 3.
Building a query in PySpark costs almost nothing. A hundred chained transformations are a hundred tiny socket messages that build a plan inside the JVM. This is exactly why the DataFrame API runs at the same speed from Python as from Scala — the engine is identical, and Python only described the work.
Why this matters for your career

The persistent myth "Scala Spark is faster than PySpark" is only true for RDDs and Python UDFs. For DataFrame and SQL work — which is what modern Spark is — the two are within noise of each other, because in both cases the JVM executes an identical optimised plan. Choose Python for the ecosystem, and just avoid the two slow paths below.

3. The executor side: Python workers

On the executor, the same split exists — but it only appears when your task actually needs to run Python code. That is the whole rule:

Pure DataFrame operations

filter, select, join, groupBy, built-in functions like upper() or datediff(). Catalyst compiles these to JVM bytecode. No Python runs on the executor at all. Zero serialization.

Python-executing operations

A @udf, rdd.map(lambda …), mapInPandas. Spark must start a Python worker process beside the executor JVM and pipe every row through it.

PATH A — pure DataFrame op (fast) Parquet partition 128 MB Executor JVM — generated bytecode filter → project → aggregate, all inline result partition 0 copies 0 Python PATH B — Python UDF (the expensive one) Executor JVM holds rows as compact Tungsten binary rows Python worker a separate OS process running your lambda ① serialize row (pickle) ④ deserialize result back into JVM ② over a local socket ③ Python computes on plain objects The bill, per row serialize + copy + deserialize ×2 + Catalyst cannot see inside → no pushdown The fix, in order of preference 1. Use a built-in function instead (Path A) · 2. If you must use Python, use a pandas_udf — Arrow moves whole batches, not rows · 3. Only then a plain @udf
Path B is where PySpark earns its bad reputation. A plain Python UDF pays serialization twice per row and blinds the optimiser. A pandas_udf keeps the same Python flexibility but moves data in Arrow batches, typically 3–50× faster. Lesson 16 measures it. 
from pyspark.sql import functions as F
from pyspark.sql.functions import udf, pandas_udf
from pyspark.sql.types import DoubleType
import pandas as pd

# ❌ SLOWEST — Path B, one round trip per row, optimiser blinded
@udf(DoubleType())
def to_celsius_udf(f):
    return (f - 32) * 5 / 9
df1 = df.withColumn("c", to_celsius_udf("temp_f"))

# ⚠️ MUCH BETTER — still Python, but Arrow moves whole batches
@pandas_udf(DoubleType())
def to_celsius_pandas(f: pd.Series) -> pd.Series:
    return (f - 32) * 5 / 9
df2 = df.withColumn("c", to_celsius_pandas("temp_f"))

# ✅ BEST — Path A, zero Python on the executor, fully optimisable
df3 = df.withColumn("c", (F.col("temp_f") - 32) * 5 / 9)
The habit that makes you good at PySpark

Before writing a UDF, spend 60 seconds searching pyspark.sql.functions. Spark ships roughly 400 built-in functions including regex, JSON parsing, date arithmetic, array transforms and aggregations. The UDF you are about to write probably already exists as a native function, and the native one will be several times faster and will survive Catalyst optimisation.

4. Apache Arrow: the fix for the boundary

Arrow is a language-neutral columnar in-memory format. Its value here is simple: the JVM and Python can both read the same Arrow buffer without converting it, so crossing the boundary becomes a bulk memory handoff instead of a per-row pickle loop.

OperationWithout ArrowWith Arrow
df.toPandas()Every row pickled individually, then a pandas DataFrame built row by rowOne columnar batch transfer; often 5–20× faster
spark.createDataFrame(pdf)Row-by-row conversion on the driverBulk columnar conversion
@pandas_udfn/a — this API only exists because of ArrowBatches of rows arrive as a pd.Series
Plain @udfRow-by-row pickleNo benefit — Arrow does not apply
 
spark = (SparkSession.builder
         .appName("arrow-on")
         .config("spark.sql.execution.arrow.pyspark.enabled", "true")
         # if a type is unsupported, silently fall back instead of failing
         .config("spark.sql.execution.arrow.pyspark.fallback.enabled", "true")
         # rows per Arrow batch — lower it if executors hit memory pressure
         .config("spark.sql.execution.arrow.maxRecordsPerBatch", "10000")
         .getOrCreate())
Caveat

Arrow does not support every Spark type equally well (deeply nested structs, some map types, MapType keys). With fallback enabled Spark quietly reverts to the slow path rather than erroring — good for reliability, but it means a job can silently lose the speedup. If a toPandas() is mysteriously slow, check your schema for exotic types.

5. Reading PySpark errors properly

Because of the two-process design, a PySpark stack trace has two halves. Most people read the wrong one. 

Traceback (most recent call last):
  File "job.py", line 42, in <module>
    df.write.parquet("/out")                 <-- ① Python half: WHERE you called from.
py4j.protocol.Py4JJavaError: An error occurred while calling o45.parquet.
: org.apache.spark.SparkException: Job aborted due to stage failure:
  Task 7 in stage 3.0 failed 4 times, most recent failure:
  Lost task 7.3 in stage 3.0 (TID 412, ip-10-0-1-42.ec2.internal, executor 5):
  org.apache.spark.api.python.PythonException:
    Traceback (most recent call last):
      File "/mnt/yarn/.../worker.py", line 619, in main
      File "job.py", line 28, in parse_date       <-- ③ the REAL cause: your UDF,
    ValueError: time data 'N/A' does not match format '%Y-%m-%d'    running on executor 5
                                                  <-- ② JVM half: usually boilerplate

How to read it in three moves

  1. Skip to the bottom. The deepest exception is the real one. Everything above is the call chain.
  2. Look for PythonException. If it is there, the bug is in your Python code running on an executor, and the nested traceback points at the exact line.
  3. "Task X failed 4 times" means Spark retried and gave up — so it is a deterministic bug in your logic, not a flaky machine. A genuinely flaky node usually succeeds on retry.
Two environment errors that waste everyone's first afternoon

1. Python version mismatch. The driver and every executor must run the same Python version.

Exception: Python in worker has different version 3.9 than that in driver 3.11,
PySpark cannot run with different minor versions.

export PYSPARK_PYTHON=/usr/bin/python3.11          # used by executors
export PYSPARK_DRIVER_PYTHON=/usr/bin/python3.11   # used by the driver

2. Missing library on executors. pip install pandas on your laptop does nothing for the executors — they are different machines. The package must exist on every executor, shipped via --py-files, a packaged virtualenv, or baked into the container image (lesson 28).

6. Spark Connect — the modern architecture

Spark 3.4 introduced Spark Connect, which replaces the Py4J subprocess with a proper client/server protocol. The client sends an unresolved logical plan over gRPC; a remote Spark server analyses, optimises and executes it, streaming results back as Arrow batches.

Classic (Py4J) Python needs a local JVM ↓ Driver JVM (local) same machine, same host Executors Java on every client machine · no upgrade isolation · heavy IDE setup Spark Connect (3.4+) Python (thin) no JVM needed ✓ gRPC: plan out, Arrow back Remote Spark Connect server driver + executors, elsewhere thin client · embeddable anywhere · client and server upgrade independently
You do not need Connect to learn Spark, and everything in this course works on both. But it is the direction of travel — it is how Databricks Connect and most notebook services work now — so knowing the distinction is worth a sentence in an interview.

Recap

  • Spark is a JVM system. PySpark is a client that drives it over a Py4J socket.
  • Only API calls and handles cross the driver bridge — never your data. That is why the DataFrame API costs nothing extra in Python.
  • Python only runs on executors when you force it to — via a UDF, an RDD lambda, or a pandas API.
  • A plain Python UDF serializes every row twice and hides the logic from Catalyst. It is the single most common cause of slow PySpark.
  • Arrow makes the crossing columnar and bulk. Enable it, and prefer pandas_udf over udf.
  • Read tracebacks bottom-up and look for PythonException to find your own bug.

Checkpoint

1 · A colleague claims your team should rewrite its pipeline in Scala because "PySpark is slow". The pipeline is entirely DataFrame operations and built-in functions. What is the accurate response?
For pure DataFrame/SQL work, Python only describes the query; Catalyst compiles it and the JVM runs identical bytecode either way. Py4J carries API calls and object handles, not rows. Rewriting in Scala is justified only if the pipeline is dominated by Python UDFs or raw RDD lambdas — and even then, a pandas_udf is usually the cheaper fix.
2 · Which of these causes a Python worker process to be started on the executors?
Filters, aggregations and joins are all compiled to JVM bytecode by Catalyst — no Python runs on the executor. Only code that must execute actual Python (a @udf, a pandas_udf, an RDD lambda, mapInPandas) forces Spark to fork a Python worker and pipe rows across a socket.
3 · Your df.toPandas() on a 2 GB result takes 6 minutes. What is the single most likely fix?
Without Arrow, toPandas() pickles each row individually and rebuilds a pandas DataFrame row by row on the driver — a notoriously slow path. Arrow transfers the same data as columnar batches, commonly 5–20× faster. (Separately: pulling 2 GB to the driver at all deserves a second look — make sure the driver has the memory for it.)