Broadcast Variables, Accumulators & Performance

Broadcast Variables & Accumulators

# Broadcast — cached ONCE on every worker, instead of re-shipped per task
country_codes = sc.broadcast({'US': 'United States', 'UK': 'United Kingdom'})

def expand_country(code):
    return country_codes.value.get(code, 'Unknown')

result = rdd.map(lambda row: (row[0], expand_country(row[1])))

# Accumulator — worker tasks can only ADD; driver reads the final total
error_count = sc.accumulator(0)

def process(record):
    global error_count
    if record.get('status') == 'error':
        error_count.add(1)
    return record

rdd.foreach(process)
print(error_count.value)  # read only AFTER the action that triggers processing

Broadcast Joins

from pyspark.sql.functions import broadcast

# Small side fits in memory -> broadcast to every worker, join happens
# LOCALLY per partition of the large side, no shuffle for the large DataFrame
large_df.join(broadcast(small_lookup_df), 'category_id')

Data Skew & Partitioning

groupByKey/reduceByKey/joins partition by key, not row count — a few disproportionately "hot" keys create oversized partitions that bottleneck the whole job while other tasks finish quickly. Salting a skewed key (appending a random suffix to spread it across synthetic sub-keys, aggregating the partial results afterward) is the standard mitigation.

df.repartition(200, 'region')   # full shuffle — can increase OR decrease partitions
df.coalesce(10)                 # cheaper — merges partitions, decrease-only, no shuffle

df.cache()   # lazy — actual caching happens on the FIRST action after this call

Python UDF Overhead

from pyspark.sql.functions import udf, pandas_udf
from pyspark.sql.types import StringType

# Plain Python UDF — crosses the Py4J bridge row-by-row, bypasses Catalyst
@udf(returnType=StringType())
def normalize(text):
    return text.strip().lower()

# Pandas UDF — batches data via Arrow, meaningfully less cross-process overhead
@pandas_udf(StringType())
def normalize_vectorized(series):
    return series.str.strip().str.lower()