PySpark Performance & TuningPySpark: производительность и тюнинг

To reason about speed you need the runtime hierarchy. One action → a job → split into stages at shuffle boundaries → each stage is many tasks, one per partition, run by executors. The driver coordinates.

Чтобы рассуждать о скорости, нужна иерархия выполнения. Одно действие → job → делится на stage'ы на границах shuffle → каждый stage — это много task'ов, по одному на партицию, выполняемых executor'ами. Driver координирует.

A shuffle redistributes rows across the cluster so that rows with the same key land on the same partition (needed for join/groupBy). It writes intermediate files to disk and sends them over the network — slow, and the most common reason a job drags.

Shuffle перераспределяет строки по кластеру так, чтобы строки с одинаковым ключом попали в одну партицию (нужно для join/groupBy). Он пишет промежуточные файлы на диск и шлёт их по сети — медленно, и это самая частая причина тормозов.

from pyspark.sql.functions import broadcast
# force a broadcast when you KNOW one side is small
big.join(broadcast(small_dim), on="dim_id", how="left")

Skew = one key has vastly more rows than others (e.g. null user_id, or a mega-customer). After a shuffle, one task gets a giant partition while the rest finish — that one straggler holds up the whole stage. Classic symptom: 199 tasks done in seconds, 1 task runs for an hour.

Перекос = у одного ключа сильно больше строк, чем у других (например null user_id или мега-клиент). После shuffle одной task достаётся огромная партиция, пока остальные закончили — этот «отстающий» держит весь stage. Классический симптом: 199 task за секунды, 1 task час.

# salting sketch: spread a hot key over N buckets
N = 16
big_salted   = big.withColumn("salt", (F.rand() * N).cast("int"))
small_salted = small.withColumn("salt", F.explode(F.array(*[F.lit(i) for i in range(N)])))
big_salted.join(small_salted, on=["key", "salt"])

Because evaluation is lazy, a DataFrame is recomputed from scratch every action. If you reuse one across multiple actions, cache it.

Поскольку вычисления ленивые, DataFrame пересчитывается с нуля при каждом действии. Если переиспользуешь его в нескольких действиях — кэшируй.

clean = expensive_pipeline(df).cache()   # = persist(MEMORY_AND_DISK)
clean.count()        # materializes the cache
a = clean.filter(...).count()
b = clean.groupBy(...).agg(...)   # reuses cache, no recompute
clean.unpersist()    # free it when done

A plain Python UDF is a black box to Catalyst: it can't optimize through it, and every row is serialized from JVM → Python and back (slow, no pushdown). Order of preference:

Обычный Python-UDF — чёрный ящик для Catalyst: он не может оптимизировать сквозь него, и каждая строка сериализуется JVM → Python и обратно (медленно, без pushdown). Порядок предпочтения:

• 1. Built-in F.* functions — always try first; fully optimized.
• 2. pandas_udf (vectorized) — uses Apache Arrow to process batches of rows; 10–100× faster than a row UDF.
• 3. Plain udf — last resort, for logic with no built-in equivalent.

• 1. Встроенные F.* — всегда пробуй первыми; полностью оптимизированы.
• 2. pandas_udf (векторизованный) — через Apache Arrow обрабатывает пачки строк; в 10–100× быстрее построчного UDF.
• 3. Обычный udf — последнее средство, для логики без встроенного аналога.

from pyspark.sql.functions import pandas_udf
import pandas as pd

@pandas_udf("double")
def norm(x: pd.Series) -> pd.Series:
    return (x - x.mean()) / x.std()   # vectorized, Arrow-backed

df.withColumn("z", norm("amount"))

df.explain(True)   # parsed → analyzed → optimized → physical plan

Look for these in the physical plan and the Spark UI:

Ищи это в физическом плане и в Spark UI:

• Exchange = a shuffle. Count them; each is expensive.
• BroadcastHashJoin vs SortMergeJoin — did broadcast actually kick in?
• PushedFilters / partition pruning — is your filter reaching the source?
• Spark UI: one task far slower than the rest → skew. Heavy spill → memory pressure / repartition needed.

• Exchange = shuffle. Считай их; каждый дорогой.
• BroadcastHashJoin vs SortMergeJoin — broadcast реально сработал?
• PushedFilters / отсечение партиций — твой фильтр доходит до источника?
• Spark UI: одна task сильно медленнее остальных → перекос. Сильный spill → нехватка памяти / нужен repartition.