44. Capstone 4 — rewrite a SQL report as a DataFrame pipeline

A teammate hands you a SQL report and asks you to fold it into a PySpark job. You'll rewrite it with the DataFrame API — and then prove the two are equivalent, closing the loop this course has been building toward.

The SQL report

Electronics revenue by shipping country, biggest first:

orders.dropDuplicates().createOrReplaceTempView("o")

spark.sql("""
    SELECT country, ROUND(SUM(quantity * unit_price), 2) AS revenue
    FROM o
    WHERE category = 'electronics'
    GROUP BY country
    ORDER BY revenue DESC
""").show()

Translate clause by clause

Each SQL clause maps to a DataFrame method:

Prove they match

Build the DataFrame version, then print its .sql() — you'll see it compiles to the same query you started from. Same plan, same numbers, two front ends. That's been true since Module 7; here you see it on your own pipeline.

df = (
    orders.dropDuplicates()
    .filter(F.col("category") == "electronics")
    .withColumn("lt", F.col("quantity") * F.col("unit_price"))
    .groupBy("country")
    .agg(F.round(F.sum("lt"), 2).alias("revenue"))
    .orderBy(F.col("revenue").desc())
)
df.show()
print(df.sql(pretty=True))

Your turn

Write the DataFrame-API equivalent of the SQL report: deduplicated orders, electronics only, revenue (sum(quantity * unit_price), rounded to 2) per country, ordered by revenue descending. Assign to result.

result = orders  # <- dedupe, filter electronics, sum(quantity*unit_price) AS revenue per country, order desc
result.show()

result = (
    orders.dropDuplicates()
    .filter(F.col("category") == "electronics")
    .withColumn("lt", F.col("quantity") * F.col("unit_price"))
    .groupBy("country")
    .agg(F.round(F.sum("lt"), 2).alias("revenue"))
    .orderBy(F.col("revenue").desc())
)

That's the course. You can read a nested source, clean it, join it, aggregate it, window it, and express any of it as SQL or DataFrame code — the working vocabulary of production PySpark. The video track picks up where the browser can't follow: how Spark runs all this across a cluster.