Spark SQL

By the end of this lesson you will be able to register a DataFrame as a view, query it with spark.sql(), mix SQL and DataFrame code in one pipeline, pass parameters safely, and save results as tables.

You already know SQL from the SQL tutorial. Spark SQL lets you use that knowledge on terabytes of data. Better still, SQL and the DataFrame API run on the same engine with the same optimiser, so neither is faster. Choose whichever is clearer for each step.

Step 1: register a view and query it

A DataFrame has no table name. To use it in SQL, register it as a temporary view: a name that points at the DataFrame for the rest of the session. No data is copied.

orders = spark.read.csv("orders.csv", header=True,
    schema="order_id INT, customer STRING, city STRING, category STRING, "
           "amount INT, status STRING, order_date DATE")
orders.createOrReplaceTempView("orders")

top_cities = spark.sql("""
    SELECT city,
           COUNT(*)    AS orders,
           SUM(amount) AS revenue
    FROM orders
    WHERE status = 'paid'
    GROUP BY city
    HAVING SUM(amount) > 1000
    ORDER BY revenue DESC
""")
top_cities.show()

+-----+------+-------+
| city|orders|revenue|
+-----+------+-------+
|Kochi|     2|  13898|
| Pune|     3|   4548|
+-----+------+-------+

spark.sql() returns an ordinary DataFrame. It is lazy like any other: the query runs only when you call an action such as show().

You can try the same SQL right here. Python's built-in sqlite3 module plays the role of the SQL engine, and pandas plays the role of the DataFrame:

import sqlite3
import pandas as pd

orders = pd.DataFrame({
    "order_id": range(1, 9),
    "city": ["Pune", "Kochi", "Jaipur", "Pune", "Lucknow", "Kochi", "Pune", "Jaipur"],
    "category": ["Books", "Electronics", "Fashion", "Electronics", "Books", "Fashion", "Fashion", "Electronics"],
    "amount": [499, 12999, 1299, 2450, 350, 899, 1599, 7999],
    "status": ["paid", "paid", "cancelled", "paid", "paid", "paid", "paid", "returned"],
})
con = sqlite3.connect(":memory:")
orders.to_sql("orders", con, index=False)     # like createOrReplaceTempView

top_cities = pd.read_sql("""
    SELECT city, COUNT(*) AS orders, SUM(amount) AS revenue
    FROM orders
    WHERE status = 'paid'
    GROUP BY city
    HAVING SUM(amount) > 1000
    ORDER BY revenue DESC
""", con)
print(top_cities)

    city  orders  revenue
0  Kochi       2    13898
1   Pune       3     4548

Step 2: mix SQL and the DataFrame API

Because spark.sql() returns a DataFrame and any DataFrame can become a view, you can switch freely. A common style is SQL for the main query and the DataFrame API for small extra steps.

from pyspark.sql import functions as F, Window

result = (
    spark.sql("SELECT category, amount FROM orders WHERE status = 'paid'")
    .groupBy("category")
    .agg(F.sum("amount").alias("revenue"))
    .withColumn("share_pct",
        F.round(F.col("revenue") * 100 / F.sum("revenue").over(Window.partitionBy()), 1))
    .orderBy(F.desc("revenue"))
)

+-----------+-------+---------+
|   category|revenue|share_pct|
+-----------+-------+---------+
|Electronics|  15449|     82.2|
|    Fashion|   2498|     13.3|
|      Books|    849|      4.5|
+-----------+-------+---------+

That is the output of result.show().

The Window.partitionBy() part computes a grand total over all rows; you will learn windows properly in Window Functions. You can also put SQL snippets inside DataFrame code with F.expr("amount * 1.18") or selectExpr("city", "amount * 1.18 AS with_gst").

Step 3: pass parameters safely

Never build SQL by gluing user input into a string with f-strings. A value such as x' OR '1'='1 would change the meaning of your query. This attack is called SQL injection. Spark supports named parameter markers: write :name in the query and pass the values separately with args.

min_amount = 1000
city = "Pune"
spark.sql(
    "SELECT order_id, customer, amount FROM orders "
    "WHERE city = :city AND amount >= :min_amount ORDER BY order_id",
    args={"city": city, "min_amount": min_amount},
).show()

+--------+--------+------+
|order_id|customer|amount|
+--------+--------+------+
|       4|   Meera|  2450|
|       7|  Farhan|  1599|
+--------+--------+------+

Spark treats the values as data, never as SQL code. This is the same idea as ? placeholders in sqlite3 or prepared statements in other databases.

Step 4: views, tables and the catalog

There are a few kinds of named objects. The differences matter in shared notebooks and scheduled jobs.

spark.sql("CREATE DATABASE IF NOT EXISTS sales")
top_cities.write.mode("overwrite").saveAsTable("sales.top_cities")
spark.sql("SHOW TABLES IN sales").show()
print(spark.catalog.tableExists("sales.top_cities"))

+---------+----------+-----------+
|namespace| tableName|isTemporary|
+---------+----------+-----------+
|    sales|top_cities|      false|
+---------+----------+-----------+

True

This output is traced, not captured: on our Windows test machine, CREATE DATABASE failed because Hadoop's Windows helper winutils.exe was missing. It works on Linux, macOS and WSL. On a laptop, managed tables are stored as Parquet files under a spark-warehouse folder next to your script, and their names are remembered only while the session runs unless a persistent catalog is configured. In companies, a shared metastore (such as the Hive Metastore or a cloud catalog like AWS Glue or Unity Catalog) remembers table names, schemas and file locations for everyone. Lakehouse table formats in Data Lakes & Lakehouses build on this.

Useful Spark SQL features

Spark SQL follows the ANSI SQL standard closely (ANSI mode is on by default in Spark 4) and adds some extras:

• CASE WHEN, CTEs with WITH, subqueries, UNION, and window functions all work as in standard SQL.
• Read files directly: SELECT * FROM parquet.\clean/orders\`` queries a Parquet folder without creating a view.
• EXPLAIN SELECT ... prints the plan, just like df.explain().
• Join hints such as SELECT /*+ BROADCAST(c) */ ... guide the optimiser (see Performance Tuning).
• The spark-sql command opens an interactive SQL shell, handy for quick checks.

Common mistakes

• Building queries with f-strings and user input. Use args with :name markers.
• Expecting a temp view to survive a restart. Save a table if others need the data later.
• Thinking SQL is slower than DataFrames (or the reverse). Both compile to the same plan; check with explain().
• Forgetting that spark.sql() is lazy. Nothing runs until an action.

Key points

• createOrReplaceTempView gives a DataFrame a name for SQL; no data is copied.
• spark.sql() returns a DataFrame, so SQL and the DataFrame API mix freely.
• SQL and DataFrame code share one optimiser and run equally fast.
• Pass values with args={...} and :name markers to avoid SQL injection.
• Managed tables (saveAsTable) store data; a metastore remembers tables across sessions.

query = """
    SELECT category,
           COUNT(*) AS paid_orders,
           SUM(amount) AS revenue,
           SUM(CASE WHEN amount >= 2000 THEN 1 ELSE 0 END) AS big_orders
    FROM orders
    WHERE status = 'paid'
    GROUP BY category
    ORDER BY revenue DESC
"""
print(pd.read_sql(query, con))

      category  paid_orders  revenue  big_orders
0  Electronics            2    15449           2
1      Fashion            2     2498           0
2        Books            2      849           0