Apache Spark / Intermediate: distribution and execution evidence / Lesson 12

Python UDFs, Arrow and worker contracts

Extend transformations only where necessary and preserve type, null, dependency and retry semantics across Python workers.

About 4 min readContent reviewed 2026-10-28

On this page

Lesson goals

Prefer supported native expressions when they express the rule.
Distinguish scalar UDF, vectorized execution and local pure-function testing.

Let the optimizer see a simple domain rule

A reading classification based on NULL and sign is directly expressible with Spark when and column comparisons. Keeping it in native expressions exposes its structure to the SQL engine and avoids unnecessary Python serialization. This is not a claim that every Python function is slow or every native expression is free; choose from semantic clarity and measured cost. The fragment assumes kwh is already typed numeric data, so it does not parse strings or validate measurement units. Keep those earlier boundaries explicit. A single function called classify can otherwise hide several incompatible contracts.

python Illustrative technical example; logical results are reasoned, not engine-tested.

from pyspark.sql import functions as F
classified = readings.withColumn('quality',
    F.when(F.col('kwh').isNull(), F.lit('missing'))
     .when(F.col('kwh') < 0, F.lit('negative'))
     .otherwise(F.lit('known')))

For typed values None,-1,0, the logical labels are missing,negative,known. No Spark expression is executed.

A Python UDF crosses a runtime boundary

A scalar Python UDF receives and returns values under a declared Spark type while executing in Python workers. The driver’s installed packages and files are not automatically present everywhere. Serialization, Python version, worker environment and data conversion become part of the interface. A closure that captures a large driver object can ship unintended data or fail serialization. Do not access a driver SparkSession from worker code or open a new remote database connection for every reading. Domain logic should be deterministic, null-aware and isolated from side effects before it is wrapped for distributed execution.

Test the local function without claiming a Spark test

The pure function below classifies only numeric values or None. Its Python behavior can be tested safely without importing PySpark. The separate udf wrapper is a platform fragment and remains non-runnable. Passing a string or an object outside the declared contract should not be silently accepted as a valid measurement. A successful local assertion proves the function’s branch logic, not Spark serialization, return-type coercion or worker packaging. Keep those verification scopes distinct in reports, especially when optional Arrow execution changes conversion behavior.

python Illustrative technical example; logical results are reasoned, not engine-tested.

def classify_numeric(value):
    if value is None:
        return 'missing'
    return 'negative' if value < 0 else 'known'

# Platform wrapper, not executed here:
classify_udf = F.udf(classify_numeric, 'string')

Only classify_numeric is eligible for isolated standard-library tests. The F.udf wrapper requires the Spark environment and is not executed.

Vectorized transport does not erase semantic contracts

Pandas UDFs process supported batches through Arrow, while Arrow-optimized scalar UDFs are a distinct documented mechanism. Batch transfer can reduce overhead but requires compatible optional dependencies and declared input/output shapes. Grouped operations can load large groups into Python memory, so a batch API does not guarantee bounded memory for every grouping pattern. Timestamp zones, nullable integers, decimal values and nested data need representative conversion tests. Do not install pandas or Arrow merely to label an unexecuted example tested. This course performs no Arrow conversion, Python-worker launch or third-party dependency installation.

Worker errors require the original cause and input scope

A Python exception can surface through a distributed task failure with a longer wrapper stack trace. Preserve the original exception type and the smallest approved synthetic input that reproduces it. A missing package is different from a return-type mismatch or an unexpected NULL. Review retry behavior before attaching logging or external requests to the function, because the same row can be evaluated more than once. Compare native and UDF results on a complete edge-case fixture before performance experiments. No engine comparison or speed claim is made here; only the local pure classification model is validated.

Common errors and fixes

A worker cannot import a driver-only package

Why it happens: Driver and executor Python environments were assumed identical.
How to fix it: Use an approved reproducible worker packaging strategy for the deployment.
Prevent it next time: Record Python and dependency compatibility separately from application source.

A local function test is reported as a distributed UDF test

Why it happens: Pure logic validation was confused with serialization and worker execution.
How to fix it: Label the tested boundary and schedule actual engine checks only in an authorized lab.
Prevent it next time: Keep local and platform test evidence separate.

A grouped vectorized function exhausts Python memory

Why it happens: A large group was treated as a safely bounded batch.
How to fix it: Inspect group sizes and choose a correct bounded processing design.
Prevent it next time: Test hot-key groups and return-shape contracts.

Check your understanding

1. What does classify_numeric return for None,-1,0?

Show worked solution

It returns missing,negative,known. These are local Python branch results for the declared input domain. They do not prove the UDF wrapper serializes or returns the expected Spark schema. Keep a separate authorized engine test when that integration is required.

2. Why prefer a native when expression for this simple sign rule?