How do I run Spark SQL from PySpark?

Summary

• Call spark.sql() on a SparkSession. It runs the query and returns the result as a DataFrame, so you can keep chaining DataFrame methods.
• To query a DataFrame by name, register it as a temporary view with createOrReplaceTempView(). Temporary views are session-scoped. A global temporary view is shared among all sessions until the application ends, and you refer to it as global_temp.<name>.
• Pass values with parameter markers: :name with a dict, or ? with a list. Named markers need Spark 3.4.0 and positional markers need Spark 3.5.0.
• In a Databricks notebook, spark already exists. Don't create your own SparkSession.

How do I run Spark SQL from PySpark?

Call spark.sql(). The Spark docs say "The sql function on a SparkSession enables applications to run SQL queries programmatically and returns the result as a DataFrame."

from pyspark.sql import SparkSession

# Outside Databricks. In a Databricks notebook, `spark` already exists.
spark = SparkSession \
    .builder \
    .appName("Python Spark SQL basic example") \
    .getOrCreate()

spark.sql("SELECT * FROM range(10) where id > 7").show()
# +---+
# | id|
# +---+
# |  8|
# |  9|
# +---+

How do I query a DataFrame with SQL?

Register the DataFrame as a view, then refer to the view by name:

df = spark.range(1, 10)  # one column named id
df.createOrReplaceTempView("numbers")

result = spark.sql("SELECT id * 2 AS doubled FROM numbers WHERE id > 5")
result.filter("doubled > 14").show()
# +-------+
# |doubled|
# +-------+
# |     16|
# |     18|
# +-------+

spark.range(1, 10) creates a DataFrame with a single LongType column named id, and the end value is exclusive. filter() accepts "a string of SQL expressions", so SQL and DataFrame code mix freely.

View type	Create with	Scope	Query as
Temporary view	createOrReplaceTempView(name)	"tied to the SparkSession that was used to create this DataFrame"	SELECT * FROM name
Global temporary view	createGlobalTempView(name)	Shared among all sessions, kept alive until the Spark application ends	SELECT * FROM global_temp.name

createOrReplaceGlobalTempView also exists. All of these view methods support Spark Connect since Spark 3.4.0. The Spark guide marks a global temporary view as "cross-session" and reads a view named people back from a new session with spark.newSession().sql("SELECT * FROM global_temp.people").show().

How do I pass values into a query?

spark.sql("SELECT * FROM range(10) WHERE id > :bound", args={"bound": 7}).show()
spark.sql("SELECT * FROM range(10) WHERE id > ?", args=[7]).show()

args takes a dict for named markers (the : prefix) or a list for positional markers (?). You can't mix the two in one query. Parameterized SQL arrived in Spark 3.4.0 and positional parameters in Spark 3.5.0. To parameterize a table name, wrap the marker in IDENTIFIER(), which the docs say allows "templating of identifiers in SQL statements without opening up the risk of SQL injection attacks".
You can also reference a DataFrame directly, with no view, using keyword arguments:

mydf = spark.createDataFrame([(1, 4), (2, 4), (3, 6)], ["A", "B"])
spark.sql("SELECT {df.A}, {df[B]} FROM {df}", df=mydf).show()

The docs call this keyword formatter feature "experimental and unstable".

How do I chain SQL and DataFrame steps?

raw = spark.read.csv("data.csv", header=True)
raw.createOrReplaceTempView("raw_data")

agg = spark.sql("SELECT category, COUNT(*) AS cnt FROM raw_data GROUP BY category")
agg.filter("cnt > 100").orderBy("cnt", ascending=False).write.mode("overwrite").parquet("output/")

For SQL fragments inside DataFrame code, selectExpr() is "a variant of select() that accepts SQL expressions", and expr(), imported from pyspark.sql.functions, lets you use SQL syntax where a column is expected: df.select("Count", expr("lower(County) as little_name")).

What are the common mistakes?

Mistake	What the docs say	Fix
Querying a global view by its bare name	"we must use the qualified name to refer it"	global_temp.name
Calling createTempView twice with one name	It throws TempTableAlreadyExistsException	createOrReplaceTempView
Mixing :name and ? in one query	It "doesn't support named and positional parameters in the same SQL query"	Pick one style
Creating a SparkSession in a Databricks notebook	"Doing so leads to inconsistent behavior"	Use the existing spark

What changes on Databricks?

• Use the existing session. An attached notebook already defines spark, along with sc and sqlContext.
• Mixing languages. A cell starting with %sql runs as SQL. Variables defined in one language are not available in the REPL of another language.
• Compute type. A notebook attached to a SQL warehouse runs SQL and Markdown cells, and a Python cell throws an error. Attach to all-purpose or serverless compute to call spark.sql().
• Photon. It is "the Databricks-native vectorized query engine" for SQL and DataFrame workloads, and Catalyst still plans your query. Photon is enabled on serverless compute and SQL warehouses, and Databricks enables it by default on classic all-purpose and jobs compute.
• Spark Connect. Serverless compute uses it. A temporary view referenced in spark.sql is analyzed lazily there, so dropping or replacing the view afterwards may fail or change the results.

FAQs

Can I run SQL without creating a temporary view?

Yes. You can query tables or table-valued functions such as range(10) directly, or pass a DataFrame with a {df} placeholder as shown above.

Can I run statements other than SELECT?

Yes. The Spark docs run CREATE TABLE and LOAD DATA through spark.sql, and the SQL reference documents INSERT INTO, for example INSERT INTO students TABLE visiting_students;.

Is spark.sql() slower than the DataFrame API?

The Spark docs say: "When computing a result, the same execution engine is used, independent of which API/language you are using to express the computation." On Databricks, "Spark DataFrames and Spark SQL use a unified planning and optimization engine, allowing you to get nearly identical performance across all supported languages on Databricks".