pyspark
PySpark is the Python API for Apache Spark, the JVM engine that splits a query into tasks and runs them in parallel across a cluster. You write DataFrame or SQL code in Python, Spark's Catalyst optimizer plans it, and the work executes on JVM executors; only your own Python code runs in separate Python worker processes next to them. The pip package bundles the Spark JARs so a laptop install works for learning, but the packaging is explicitly aimed at driving an existing cluster (standalone, YARN, Kubernetes, or a Spark Connect server) rather than standing one up.
Still the default when the data genuinely does not fit on one machine, and unavoidable if your platform already runs Spark. If you are reaching for it on a single node out of habit, Polars or DuckDB will start faster, cost less to operate, and be far easier to debug.
Use it if
- Your data does not fit on one machine and you need a cluster to do the scan, join, or shuffle
- You are on Databricks, EMR, Dataproc, or Fabric, where Spark is the runtime you were handed and PySpark is how Python talks to it
- You want one engine covering batch SQL, Structured Streaming, and MLlib over the same Parquet, Iceberg, or Delta tables
- Your team already writes Spark SQL and you want Python that plans through the same optimizer instead of a second execution model
- Your data fits in memory on one box: Polars or DuckDB start in milliseconds and will finish the same aggregation while a local SparkSession is still booting a JVM
- You do not want a JVM in your stack: PySpark needs a JDK on PATH, opens a Py4J gateway, and surfaces failures as Java stack traces wrapped in Python exceptions
- You need low-latency answers: Spark optimizes throughput, and job submission plus query planning make it the wrong tool behind an HTTP request
- You rely on pandas API on Spark: since 4.0 it raises an exception when ANSI mode is on, and ANSI mode is now the default, so you must turn one of them off
- Debugging cost matters to you: a broken Python UDF comes back as a multi-page executor traceback, and shuffle or skew bugs frequently cannot be reproduced at laptop scale
Setup reality
pip install pyspark downloads the Spark JARs, so it is a large install, and it still does nothing until a JDK is on PATH. Anything past a toy DataFrame needs extras: pyspark[sql] for pandas and PyArrow, pyspark[connect] for the gRPC client. Spark 4.2 raised the PyArrow floor to 18.0.0, pandas to 2.2.0, and dropped PyPy support. If you point a pip-installed PySpark at a standalone cluster, the README warns the versions including the minor must match or you get odd errors. Windows also wants winutils.exe. Setting PYSPARK_PYTHON so the driver and executors run the same interpreter is the most common fix for the 'Python worker exited unexpectedly' failures new users hit.
Patterns
Create a SparkSessioncreate-session
from pyspark.sql import SparkSession
spark = (
SparkSession.builder
.appName("etl")
.master("local[*]")
.config("spark.sql.shuffle.partitions", "8")
.getOrCreate()
)getOrCreate() returns the session that already exists in the process, so config() calls on a second builder are silently ignored. The 200-partition shuffle default is far too high for local runs.
Read Parquet and CSV with an explicit schemaread-files
df = spark.read.parquet("s3a://bucket/events/")
orders = (
spark.read
.option("header", True)
.schema("user_id STRING, order_id STRING, ts TIMESTAMP, amount DOUBLE")
.csv("data/orders.csv")
)inferSchema costs an extra full pass over the data; a DDL schema string skips it and stops types drifting between runs. In Spark 4.2, .option(key, None) means "unset" instead of forwarding a Java null.
Filter, derive, and select columnsselect-filter
from pyspark.sql import functions as F
recent = (
orders
.filter(F.col("ts") >= F.lit("2026-01-01"))
.withColumn("amount_usd", F.col("amount") * F.lit(1.09))
.select("user_id", "order_id", "amount_usd")
)Spark 4.0 removed DataFrame, Column, and StructType from the wildcard import of pyspark.sql.functions, so import those from pyspark.sql and pyspark.sql.types instead.
Group and aggregategroup-aggregate
totals = (
orders.groupBy("user_id")
.agg(
F.sum("amount").alias("total"),
F.count_distinct("order_id").alias("orders"),
F.max("ts").alias("last_seen"),
)
)Without alias() you get column names like sum(amount) that need backticks everywhere downstream. count_distinct is the current name; countDistinct still works as an alias.
Join, and broadcast the small sidejoin-dataframes
joined = orders.join(users, on="user_id", how="left")
# force a broadcast when the planner has no stats
joined = orders.join(F.broadcast(users), on="user_id", how="left")on="user_id" collapses the duplicate key into one column; on=orders.user_id == users.user_id keeps both and every later select("user_id") fails as ambiguous.
Take the latest row per key with a windowwindow-functions
from pyspark.sql.window import Window
w = Window.partitionBy("user_id").orderBy(F.col("ts").desc())
latest = (
orders.withColumn("rn", F.row_number().over(w))
.filter(F.col("rn") == 1)
.drop("rn")
)Leave out partitionBy and Spark moves every row into a single partition on one executor, which is the classic way to hang an otherwise fine job.
Run SQL against a DataFramesql-view
orders.createOrReplaceTempView("orders")
top = spark.sql("""
SELECT user_id, SUM(amount) AS total
FROM orders
GROUP BY user_id
ORDER BY total DESC
LIMIT 10
""")Temp views are scoped to the session that made them. Use createOrReplaceGlobalTempView and query global_temp.orders if another session has to see it.
Write a Python UDFpython-udf
from pyspark.sql.types import StringType
@F.udf(returnType=StringType())
def domain(email):
if not email:
return None
return email.split("@")[-1]
users = users.withColumn("domain", domain(F.col("email")))Spark 4.2 makes regular Python UDFs Arrow-optimized by default via spark.sql.execution.pythonUDF.arrow.enabled. A built-in function is still faster, because a UDF is opaque to Catalyst and blocks predicate pushdown.
Vectorize with a pandas UDFpandas-udf
import pandas as pd
from pyspark.sql.functions import pandas_udf
@pandas_udf("double")
def to_fahrenheit(c: pd.Series) -> pd.Series:
return c * 9 / 5 + 32
readings = readings.withColumn("temp_f", to_fahrenheit("temp_c"))Since Spark 4.2 a nullable integer column containing nulls arrives as a pandas nullable dtype (Int64) rather than float64, so UDF bodies written against the old behavior need checking.
Write partitioned Parquetwrite-partitioned
(
orders.repartition("dt")
.write
.mode("overwrite")
.partitionBy("dt")
.parquet("s3a://bucket/curated/orders/")
)partitionBy on a high-cardinality column makes one directory per value and can produce millions of tiny files. repartition on the same column first keeps the file count per partition sane.
Connect to a remote Spark Connect serverspark-connect
# pip install "pyspark[connect]"
from pyspark.sql import SparkSession
spark = SparkSession.builder.remote("sc://spark-host:15002").getOrCreate()
spark.range(5).show()A Connect session has no SparkContext, so spark.sparkContext and df.rdd raise instead of working. Code that reaches into the RDD API has to be rewritten against DataFrames first.
Pull results back to the drivercollect-results
pdf = orders.limit(1000).toPandas()
first_five = orders.take(5)
row_count = orders.count()toPandas() materializes every row in driver memory, so without a limit it will kill the driver no matter how big the cluster is. count() triggers a full job, which is not free inside a loop.
Alternatives
| Package | Registry | Pick it when |
|---|---|---|
| polars | PyPI | Single machine, dataframe-shaped work, and you want a multi-threaded engine with no JVM and no cluster. |
| duckdb | PyPI | You mostly write SQL over Parquet or CSV files and one process can hold the working set. |
| dask | PyPI | You want distributed Python that stays in Python, especially for array and pandas-shaped workloads instead of SQL. |