UDFs, Broadcast Variables and Accumulators

What it is

A user-defined function (UDF) lets you apply your own Python logic to column values when no built in function does the job. Spark also has two shared variable types for code that runs on many executors: broadcast variables share read only data, and accumulators collect counters back to the driver.

The exam asks how to create and call a UDF, why built in functions or vectorized UDFs are preferred, and which shared variable fits a requirement.

How it works

Creating and invoking a UDF

from pyspark.sql import functions as F
@F.udf("string")
def tier(score):
    return None if score is None else ("high" if score > 80 else "low")
df.withColumn("tier", tier("score"))

The return type is the second argument of F.udf, or a decorator argument, and it defaults to string if you omit it. If the function returns a value that does not match the declared type, the result is usually null instead of an error. Handle None inputs inside the function, because nulls reach it as None.

To call a UDF from SQL text, register it by name.

spark.udf.register("tier_sql", tier)
spark.sql("SELECT tier_sql(score) FROM scores")

A function decorated with @F.udf can be used in the DataFrame API directly, but it needs spark.udf.register before SQL can see it.

Why UDFs cost more

Use a built in function whenever one exists. When custom logic is unavoidable, a pandas UDF with type hints is the faster choice.

Stateful processing and StateStores

A normal UDF is stateless. It sees one row, or one batch, and keeps nothing between calls. You cannot rely on a Python global variable to carry information across rows, because rows are processed in many tasks on many executors.

For arbitrary stateful logic on streaming data, PySpark offers applyInPandasWithState, used after groupBy on a streaming DataFrame. Spark keeps the state for each key in a StateStore on the executors. State is saved with each micro batch to the checkpoint location, so it survives a restart. A timeout setting lets you expire state for idle keys.

Stateful operators such as streaming aggregations and deduplication use the same StateStore mechanism.

Broadcast variables

A broadcast variable ships a read only value to each executor once, instead of with every task.

lookup = spark.sparkContext.broadcast({"US": "United States"})
name = F.udf(lambda c: lookup.value.get(c), "string")

Read it with .value inside tasks. Use it for lookup tables used in UDFs or RDD code. It is different from a broadcast join, which is a DataFrame join strategy.

Accumulators

An accumulator is a variable that tasks can only add to and that only the driver can read.

bad = spark.sparkContext.accumulator(0)
def check(x):
    if x is None:
        bad.add(1)

Tasks call add, and the driver reads bad.value after an action. Updates made inside transformations may be applied more than once if a task is retried or a stage re-runs. Only updates inside actions, such as foreach, are guaranteed to be applied exactly once.

How the exam asks about this

• "A UDF returns null for every row although the function looks right." Check that the returned Python type matches the declared return type.
• "Count malformed records during a job and read the count on the driver." Use an accumulator.
• "Every task needs the same 50 MB dictionary; avoid sending it with each task." Use a broadcast variable.