Spark runs SQL over DataFrames, and the two APIs are interchangeable. Learn to query with SQL, join datasets, and use broadcast joins to make small-to-large joins fast.
Register a DataFrame as a temporary view and you can query it with full SQL; conversely, DataFrame operations compile to the same optimized plan as the equivalent SQL. They are two front-ends to one engine, so you mix them freely — SQL for readable set logic, the DataFrame API for programmatic pipelines. Use whichever is clearer for the task.
df = spark.read.parquet("sales.parquet")
df.createOrReplaceTempView("sales")
# full SQL over the DataFrame:
top = spark.sql("""
SELECT customer_id, SUM(amount) AS total
FROM sales
WHERE amount > 0
GROUP BY customer_id
ORDER BY total DESC
LIMIT 10
""")
top.show()
# identical result via the DataFrame API — same optimized plan underneath.
# Pick whichever reads better; you can switch mid-pipeline.Joins combine datasets on a key and are everyday work in data engineering — enriching events with dimensions, stitching sources together. The API takes the other DataFrame, the join condition, and the join type (inner, left, etc.). Joins, like aggregations, shuffle data so matching keys land together, which makes them one of the most expensive and most tuned operations.
orders = spark.read.parquet("orders.parquet")
customers = spark.read.parquet("customers.parquet")
enriched = orders.join(
customers,
on="customer_id", # or a condition: orders.cid == customers.id
how="left", # inner | left | right | outer | anti | semi
)
enriched.select("order_id", "amount", "customer_segment").show()
# a join shuffles BOTH sides so equal keys meet on the same executor —
# expensive on big data. The next topic removes the shuffle when one
# side is small.The classic data-engineering join is a huge fact table against a small dimension. Shuffling the giant side is wasteful — instead, broadcast the small table: Spark ships a copy to every executor so each can join its fact partitions locally, with no shuffle. This single technique turns many slow joins fast; Spark often does it automatically, and you can force it with a hint.
from pyspark.sql.functions import broadcast
# huge fact (billions) JOIN small dim (thousands):
result = big_facts.join(broadcast(small_dim), on="product_id", how="left")
# broadcast() sends the small table to every executor -> NO shuffle of
# the big table. Each partition joins locally. Massive speedup.
# Spark auto-broadcasts tables under spark.sql.autoBroadcastJoinThreshold
# (default 10MB). Raise it or hint broadcast() when you know a side is small.Window functions compute across a set of rows related to the current row without collapsing them — running totals, rankings, row-over-row comparisons. They are essential for analytics (e.g. “each customer’s 3rd order,” “running revenue”) and work the same as in SQL warehouses. Note that an unpartitioned window pulls all data to one executor, so always partition.
from pyspark.sql import Window
from pyspark.sql import functions as F
w = Window.partitionBy("customer_id").orderBy("ordered_at")
ranked = (orders
.withColumn("order_seq", F.row_number().over(w)) # 1st, 2nd order...
.withColumn("running_total", F.sum("amount").over(w)) # cumulative spend
.withColumn("prev_amount", F.lag("amount").over(w))) # previous order
ranked.show()
# ALWAYS partitionBy — an unpartitioned window gathers ALL rows onto one
# executor (a single-partition shuffle) and won't scale.