The DataFrame API is how you actually shape data in Spark — select, filter, add columns, and aggregate. It reads like pandas but runs distributed and lazy across the cluster.
The core verbs mirror SQL and pandas: select picks columns, filter (or where) keeps rows, and withColumn derives new columns. Because these are lazy transformations, you chain them into a pipeline that Spark optimizes as a whole. Column expressions use functions from pyspark.sql.functions rather than Python operations on scalars, because they run distributed.
from pyspark.sql import functions as F
df = spark.read.parquet("sales.parquet")
clean = (
df
.filter(F.col("amount") > 0) # keep valid rows
.select("order_id", "customer_id", "amount", "ordered_at")
.withColumn("amount_usd", F.col("amount") / 100) # derive a column
.withColumn("order_date", F.to_date("ordered_at"))
.withColumnRenamed("customer_id", "cust_id")
)
clean.show(5)
# use F.col(...) + functions (not raw Python) so it runs on the cluster.groupBy plus agg is how you collapse many rows into summaries — counts, sums, averages per key. This is where Spark earns its keep, aggregating billions of rows in parallel. Aggregations trigger a shuffle (data moves between executors so equal keys meet), which is the main cost to be aware of and the subject of the performance lesson.
from pyspark.sql import functions as F
revenue = (
clean.groupBy("cust_id")
.agg(
F.sum("amount_usd").alias("total_spent"),
F.count("*").alias("num_orders"),
F.avg("amount_usd").alias("avg_order"),
F.max("order_date").alias("last_order"),
)
.orderBy(F.desc("total_spent"))
)
revenue.show()
# groupBy triggers a SHUFFLE (rows with the same key gather on one
# executor). It's powerful but the main cost lever — see the perf lesson.A DataFrame has a schema — column names and types — and getting it right avoids silent bugs and slow inference. On big data you supply an explicit schema instead of letting Spark scan the file to infer one, which is both faster and safer. Inspecting the schema and query plan is a routine part of Spark development.
from pyspark.sql.types import StructType, StructField, StringType, DoubleType, TimestampType
schema = StructType([
StructField("order_id", StringType(), False),
StructField("customer_id",StringType(), False),
StructField("amount", DoubleType(), True),
StructField("ordered_at", TimestampType(),True),
])
df = spark.read.schema(schema).csv("orders.csv", header=True)
df.printSchema() # inspect columns + types
df.explain() # see the physical plan Spark will run
# explicit schema > inferSchema on big data: no extra scan, no wrong guesses.When built-in functions cannot express your logic, a UDF runs arbitrary Python per row. Prefer built-ins: a Python UDF forces data out of Spark’s optimized engine into the Python interpreter row by row, which is dramatically slower and invisible to the optimizer. Reach for UDFs only when necessary, and prefer pandas/vectorized UDFs when you must.
Prefer built-ins — Built-in pyspark.sql.functions run in Spark’s optimized JVM engine. A Python UDF serializes every row to Python — often 10x+ slower and opaque to the optimizer. Use UDFs only when no built-in works.from pyspark.sql import functions as F
from pyspark.sql.types import StringType
# Built-in first — fast, optimizable:
df = df.withColumn("tier",
F.when(F.col("amount_usd") > 100, "high").otherwise("low"))
# UDF only if truly needed — slow (row-by-row Python), optimizer-opaque:
@F.udf(returnType=StringType())
def classify(x):
return "high" if x and x > 100 else "low"
df = df.withColumn("tier", classify("amount_usd"))
# ^ correct but slower. Prefer F.when / built-ins; if you must UDF,
# use a pandas_udf (vectorized) over a plain Python UDF.