Definition
- Serialization cost – Spark's JVM engine must serialize each row, send it to a Python worker process, deserialize it, run the function, and serialize the result back: enormous per-row overhead.
- Optimizer opacity – Catalyst treats the UDF as a black box and cannot push filters, prune columns, or generate fused bytecode through it.
- Built-in functions – run entirely on the JVM with full Catalyst optimization; always fastest.
- Pandas UDFs – transfer data in batches via Apache Arrow; 10x to 100x faster than plain UDFs.
- Plain Python UDFs – last resort for genuinely row-wise Python logic with no built-in or vectorized equivalent.
The bank's main ledger is kept by a fast JVM clerk, but you insist every single transaction be mailed to a Python specialist in another building. The specialist opens the envelope, does one calculation, reseals it, and mails it back: one letter per transaction. The postage (serialization) and round trips dwarf the actual work. A built-in function is the JVM clerk doing it himself. A Pandas UDF mails a whole box of transactions at once instead of one letter each.
Understanding the Concepts
Why are plain Python UDFs slow?
Two independent reasons make plain Python UDFs the slowest option in PySpark.
Reason 1: Per-row serialization. Spark's engine runs on the JVM and DataFrame columns live in efficient Tungsten binary format. A Python UDF cannot run on the JVM, so Spark launches Python worker processes. For every row, the value is serialized (pickled), sent over a socket to Python, deserialized, passed to your function, and serialized back. This per-row round trip dominates the cost on large data.
Reason 2: Optimizer opacity. Catalyst understands built-in functions and can reorder, combine, push filters, and generate fused JVM bytecode. A Python UDF is a sealed box: Catalyst must materialize its inputs, call it, and cannot optimize across it.
from pyspark.sql import SparkSession, functions as F
from pyspark.sql.types import DecimalType
spark = SparkSession.builder.appName("NBI-UDF-Demo").getOrCreate()
txns = spark.read.parquet("s3://nbi-data-lake/transactions/")
# SLOW: plain Python UDF (per-row serialization + opaque)
@F.udf(DecimalType(18, 2))
def compute_fee_udf(amount):
if amount is None:
return None
return amount * 0.001
result_slow = txns.withColumn("fee", compute_fee_udf(F.col("amount")))
What is the performance hierarchy?
The hierarchy is clear: built-in functions > Pandas UDFs > plain Python UDFs.
| Approach | Where it runs | Optimization | Relative speed |
|---|---|---|---|
| Built-in functions | JVM (Tungsten) | Full Catalyst + codegen | 1x (baseline, fastest) |
| Pandas UDFs | Python (batched via Arrow) | Partial (batch amortizes cost) | 3x to 10x slower than built-in |
| Plain Python UDFs | Python (per-row via pickle) | None (opaque to Catalyst) | 10x to 100x slower than built-in |
# FASTEST: built-in function (stays on JVM)
result_fast = txns.withColumn("fee", F.col("amount") * F.lit(0.001))
# MEDIUM: Pandas UDF (batched via Arrow)
import pandas as pd
@F.pandas_udf(DecimalType(18, 2))
def compute_fee_pandas(amount: pd.Series) -> pd.Series:
return amount * 0.001
result_medium = txns.withColumn("fee", compute_fee_pandas(F.col("amount")))
# SLOWEST: plain Python UDF (per-row serialization)
result_slow = txns.withColumn("fee", compute_fee_udf(F.col("amount")))
How do built-in functions avoid the problem?
Built-in functions like F.col, F.when, F.upper, F.sum run entirely on the JVM. Catalyst understands them, can push filters to the data source, prune unused columns, and fuse operations into generated bytecode.
# Built-in: Catalyst pushes the filter to the Parquet reader
# and generates fused bytecode for the entire pipeline.
result = txns \
.filter(F.col("amount") > 1000) \
.withColumn("fee", F.col("amount") * F.lit(0.001)) \
.select("accountId", "fee")
result.explain("formatted")
# Shows: PushedFilters: [GreaterThan(amount, 1000)]
# Shows: WholeStageCodegen (fused pipeline)
== Physical Plan ==
*(1) Project [accountId, (amount * 0.001) AS fee]
+- *(1) Filter (amount > 1000)
+- FileScan parquet [...]
PushedFilters: [GreaterThan(amount,1000)]
How do Pandas UDFs reduce the overhead?
Pandas UDFs (vectorized UDFs) transfer data in batches using Apache Arrow, which is a columnar in-memory format. Instead of serializing one row at a time, Spark sends a whole batch of rows as an Arrow buffer. The UDF receives a pandas Series and returns a pandas Series.
import pandas as pd
from pyspark.sql.functions import pandas_udf
@pandas_udf("decimal(18,2)")
def risk_score(amount: pd.Series, txn_count: pd.Series) -> pd.Series:
"""Custom risk scoring that needs Python logic"""
base_score = amount / 10000
volume_factor = txn_count.clip(upper=100) / 100
return (base_score * volume_factor).round(2)
# The data crosses the JVM-Python boundary in BATCHES, not per-row
result = txns.withColumn(
"risk",
risk_score(F.col("amount"), F.col("txnCount"))
)
How does the UDF block Catalyst optimization?
A plain UDF is opaque to Catalyst. It cannot see inside, so it cannot push filters past the UDF or prune columns the UDF might reference.
# WITHOUT UDF: Catalyst pushes the filter to the Parquet reader
txns.filter(F.col("amount") > 1000) \
.withColumn("fee", F.col("amount") * F.lit(0.001))
# PushedFilters: [GreaterThan(amount, 1000)] ← pushed down!
# WITH UDF: Catalyst cannot push filter past the UDF
txns.withColumn("fee", compute_fee_udf(F.col("amount"))) \
.filter(F.col("fee") > 1.0)
# PushedFilters: [] ← NOT pushed down! Full scan required.
What are common mistakes with UDFs?
The most common mistake is using a UDF for logic that built-in functions can handle.
# WRONG: Python UDF for simple string operations
@F.udf("string")
def upper_name(name):
return name.upper() if name else None
result = txns.withColumn("upper", upper_name(F.col("name")))
# RIGHT: built-in function (stays on JVM, fully optimized)
result = txns.withColumn("upper", F.upper(F.col("name")))
# WRONG: Python UDF for conditional logic
@F.udf("string")
def classify(amount):
if amount > 100000: return "HIGH"
if amount > 10000: return "MEDIUM"
return "LOW"
# RIGHT: built-in when/otherwise
result = txns.withColumn("class",
F.when(F.col("amount") > 100000, "HIGH")
.when(F.col("amount") > 10000, "MEDIUM")
.otherwise("LOW")
)
When is a plain Python UDF acceptable?
Use a plain UDF only when you need row-level Python logic that has no built-in or Pandas UDF equivalent.
# Acceptable: calling an external Python library per-row
import hashlib
@F.udf("string")
def mask_account(account_id):
"""PII masking using hashlib: no built-in equivalent"""
if account_id is None:
return None
return hashlib.sha256(account_id.encode()).hexdigest()[:12]
result = txns.withColumn("masked_id", mask_account(F.col("accountId")))
# This genuinely needs Python. Use a Pandas UDF if possible.
What should a senior engineer know?
A senior engineer rewrites plain UDFs to built-ins whenever possible, uses Pandas UDFs for custom Python logic, and only resorts to plain UDFs for genuinely row-wise Python that has no alternative.
# Senior decision tree:
# 1. Can I express this with built-in functions? → Use built-in
# 2. Can I express this as a batch operation on pandas Series? → Pandas UDF
# 3. Do I need a Python library per-row? → Plain UDF (last resort)
# Also: enable Arrow for Pandas UDFs
spark.conf.set("spark.sql.execution.arrow.pyspark.enabled", "true")
Visual Overview
Plain Python UDF (per-row, slow):
JVM Executor Python Worker
┌──────────┐ serialize ┌──────────┐
│ Row 1 │────────────→│ udf(row1)│
│ │←────────────│ result1 │
│ Row 2 │────────────→│ udf(row2)│
│ │←────────────│ result2 │
│ ... │ N round │ ... │
│ Row N │ trips! │ udf(rowN)│
└──────────┘ └──────────┘
Pandas UDF (batched via Arrow, fast):
JVM Executor Python Worker
┌──────────┐ Arrow batch ┌──────────┐
│ Rows │─────────────→│ pandas │
│ 1..10000 │ │ Series │
│ │←─────────────│ result │
└──────────┘ 1 round trip└──────────┘
Built-in function (no boundary crossing):
JVM Executor
┌──────────────────────┐
│ Tungsten codegen │
│ col("amount") * 0.001│
│ (all on JVM, fast!) │
└──────────────────────┘
Key Takeaway
Plain Python UDFs are slow in PySpark for two reasons. First, per-row serialization: every row crosses the JVM-to-Python process boundary, which involves serializing, sending over a socket, deserializing, running the function, and serializing back. Second, optimizer opacity: the UDF is a black box to Catalyst, blocking filter pushdown, column pruning, and code generation.
The performance hierarchy is clear: built-in Spark functions (JVM-native, fully optimized) are fastest, Pandas UDFs (batched via Arrow) are 10x to 100x faster than plain UDFs, and plain Python UDFs (per-row serialization) are the last resort.
At the National Bank of India, a developer computed a transaction fee with a plain Python UDF and the job crawled on millions of rows. I rewrote it as F.col("amount") * F.lit(0.001), which stayed entirely on the JVM and ran almost instantly. For a custom risk score that genuinely needed Python, I used a Pandas UDF with Arrow enabled and recovered most of the speed.
The rule is: express it with built-ins if you can, use a Pandas UDF when you need real Python, and only write a plain UDF when neither will do.
Remember one sentence: built-in functions run on the JVM with full optimization, Pandas UDFs batch via Arrow, and plain Python UDFs serialize per-row, so always prefer built-ins first.
---