Running SQL Queries on DataFrames

Spark tutorial · PySpark.in

Running SQL Queries on DataFrames

Now that we know how to register DataFrames as tables or views, we can explore running SQL queries directly on these views using PySpark. The spark.sql() function is the workhorse for this  it allows you to execute a SQL query as a string and returns the result as a new DataFrame. This means you can leverage the full expressiveness of SQL (SELECT, WHERE, GROUP BY, HAVING, JOIN, etc.) on your distributed data.

Using Spark SQL queries in PySpark typically follows these steps:

1. Create or load a DataFrame (using spark.read or from an existing RDD or collection).
2. Register the DataFrame as a temporary view with a name.
3. Call spark.sql("SQL QUERY") to execute your query string, referring to the view name in the FROM clause.
4. Get back a DataFrame as the query result, which you can further .show(), transform, or save.

Let’s walk through an example continuing with our people data from earlier. First, ensure the DataFrame is registered as a temp view:

Now we can run various SQL queries against the “people” view. Spark SQL supports all the standard clauses, such as SELECT, WHERE, GROUP BY, ORDER BY, etc. Here are a few examples:

Basic SELECT: Retrieve all records and columns (similar to df.show() but via SQL):

Each row of the DataFrame is now accessible in SQL. The result_df is a new DataFrame (the result of the query), so you could use it programmatically as well. For instance, you could call result_df.where("age >= 21") or any DataFrame operation on it if you wanted.

Filtering with WHERE: You can filter rows using a WHERE clause just like in any SQL:

Notice that we can choose to select specific columns (name, age) instead of * as needed, and the WHERE clause restricts the rows.

Aggregations and GROUP BY: Spark SQL lets you perform aggregations like counting, summing, averaging, etc., combined with grouping. For example, let’s count how many people fall into each age value:

This will group the records by age and count how many entries per age. Based on our small dataset:

Each age appears once (since all are distinct in our tiny set). If there were duplicates, their counts would reflect that. You could also use other aggregate functions like SUM, AVG (average), MIN, MAX similarly. (We will explore more SQL functions in the next section.)

Sorting results: To sort the output, use ORDER BY. For example, to list people by descending age:

showing the rows sorted by age in descending order.

Limiting results: If you only want the first N results, use LIMIT. For example:

would show only 2 rows out of the 3 (an arbitrary two, since no explicit ORDER BY was given).

spark.sql() returns a DataFrame: It’s worth emphasizing that spark.sql() yields a DataFrame object representing the result set of the query. You can use this result DataFrame as you would any other – e.g., call .collect() to retrieve it to the driver, write it to storage, or even create a new temp view from it. This means you can chain SQL with further PySpark DataFrame operations if needed.

For example:

This would first use SQL to get under-30 folks, then use the DataFrame API to further filter those whose name starts with J.

Under the hood: When you run spark.sql(), Spark uses the Catalyst optimizer to plan the query, just as it would with DataFrame code. The same optimizations apply (for example, the filter in WHERE is pushed down to where the data resides if possible, aggregations use combiners, etc.). The Spark documentation highlights that running SQL queries programmatically via spark.sql returns the result as a DataFrame, unifying the concept of tables and DataFrames.

In summary, Spark SQL queries on DataFrames allow you to treat your distributed data as if it were in a SQL database. You get the expressiveness of SQL and the scalability of Spark. This can be extremely powerful for teams that are accustomed to SQL or for quickly writing complex queries that might be less straightforward in the DataFrame API. You can always achieve the same with DataFrame operations, but often SQL can be more concise for certain tasks (and vice versa). Spark gives you the freedom to use either approach, and because of the underlying architecture, you don’t sacrifice performance for using SQL.

SQL Functions (SUM, AVG, CASE WHEN, etc.)

Overview: Spark SQL comes with a rich library of built-in functions for data manipulation, similar to those found in traditional SQL databases. These include aggregate functions (for summarizing data), arithmetic and mathematical functions, string functions for text processing, date/time functions, conditional expressions, and more. Leveraging these functions can significantly simplify your queries and often yields better performance than writing custom logic. In this section, we’ll cover some of the most commonly used functions in Spark SQL and how to use them in your queries.

Spark SQL’s built-in functions span many categories. Here are some key types of functions and examples of each:

• Aggregate Functions: Perform calculations across multiple rows (often with a GROUP BY). Common examples: COUNT(), SUM(), AVG() (average), MIN(), MAX(). These are used in the SELECT list typically in conjunction with grouping. For example, SELECT COUNT(*) FROM table returns the number of rows in the table. SELECT category, SUM(amount) FROM sales GROUP BY category gives the total sales per category.
• Scalar Functions: Operate on individual values (row by row). These include:
  • String functions: e.g. UPPER(str) to convert to uppercase, LOWER(str) for lowercase, LENGTH(str) to get string length, CONCAT(str1, str2) to concatenate, SUBSTR(str, pos, len) (substring) and many others for manipulating text.
  • Date and time functions: e.g. CURRENT_DATE() to get today’s date, YEAR(date)/MONTH(date)/DAY(date) to extract parts of a date, DATEDIFF(date1, date2) to get difference in days, DATE_ADD(date, interval) to add days, date_format(date, format) to format dates, etc.
  • Conditional functions/expressions: The most notable is CASE WHEN for branching logic. This allows you to derive new values based on conditions (similar to if-else). There’s also COALESCE(x,y,...) which returns the first non-null value, and various null-checking functions (NVL, etc., depending on Spark version).
  • Numerical functions: e.g. ABS(x) for absolute value, ROUND(x, d) to round to d decimal places, POWER(x, p) or the POW alias for exponents, etc.
• Miscellaneous: There are also array/map functions, JSON functions, grouping sets, window functions (for advanced aggregations over partitions of data), and more. We won’t cover those advanced ones here, but be aware Spark SQL is quite feature-rich.

Using these functions in Spark SQL is straightforward – you call them in the SQL query string just like in standard SQL. Let’s illustrate some of them with examples, using our people view and a few other hypothetical examples:

Aggregate Functions Example: Suppose we have a view sales with schema (category, amount). We can aggregate the data:

# Example: Sum of amounts per category (assuming a 'sales' view exists)

# This is a hypothetical example for illustration

This query would produce output with one row per category, showing the sum of amount, average amount, and count of records in each category. Spark’s implementation of these aggregates will ignore nulls by default (e.g., SUM of values that include NULL will treat NULL as 0 in the sum, effectively summing non-nulls. Also, note that if you use any aggregate (like SUM or AVG) and also select non-aggregated columns, you must use a GROUP BY on those non-aggregated columns (just like standard SQL).

As a concrete example with our data, let’s use an aggregate on the people view. Maybe we want to know the average age of people in our dataset:

indicating the average age is 26 (since (29+30+19)/3 = 26), total age is 78, and count is 3. (Spark will likely return avg_age as a double with one decimal as shown above.)

CASE WHEN (Conditional Expressions): The CASE WHEN expression in Spark SQL allows you to perform if-then-else logic within a query. It’s extremely useful for categorizing or transforming data based on conditions. The syntax is:

```

hide

CASE

WHEN condition1 THEN value1

WHEN condition2 THEN value2

ELSE value_else

END

```

You can have multiple WHEN clauses and an optional ELSE. Spark SQL’s CASE works just like SQL standard.

For example, let’s classify people as 'Adult' or 'Minor' based on their age:

In this example, all three are labeled "Adult" because they’re all 18 or over (if one had been under 18, they’d show "Minor"). If any age were NULL, the condition age >= 18 would be evaluated as FALSE (or unknown), so the result would fall into the ELSE and label as "Minor". You could add additional WHEN clauses for more categories if needed (e.g., different age brackets). The resulting column age_group is a derived categorical column based on the logic.

Another use of CASE is in aggregation with conditional sums or counts (sometimes known as “conditional aggregation”). For instance, you could combine CASE with SUM to count how many people are adults vs minors:

This query uses CASE inside SUM to effectively count rows meeting a condition (since it adds 1 for each row that matches, 0 otherwise). Spark SQL supports this pattern, which can be very powerful for creating pivot-table like summaries in one query.

String Functions Example: Let’s demonstrate some string operations using Spark SQL. Imagine we want to standardize the format of names to uppercase and also find the length of each name. We can do:

We see that UPPER(name) converted each name to uppercase. LENGTH(name) returned the number of characters in each name. Spark also has LOWER(name) to do the inverse (lowercase everything). Another common string function is CONCAT, which concatenates multiple strings (or columns). For example:

would produce sentences like “Michael is 29 years old”. Behind the scenes, Spark’s string functions are optimized in Scala/Java, so they execute efficiently across the dataset.

Date Functions Example: Working with dates and timestamps is another common need. Spark SQL provides many functions to extract or manipulate date/time values. Suppose we have a view events with a column event_date of type date. We could do:

Here YEAR(), MONTH(), and DAY() pulled out parts of the date. Spark also has date_format(date, 'pattern') which can format dates as strings, and functions like current_date(), current_timestamp(), datediff(endDate, startDate) (difference in days), add_months(date, n), etc. You can also convert strings to dates with to_date(string, format) if your dates are in text form.

Important Note on using functions: When using Spark SQL functions:

• The function names are case-insensitive (you can write upper or UPPER).
• If a function name is also a column name or something, you might need to disambiguate, but usually not an issue with these standard ones.
• Many functions (like the aggregates and math functions) will return null if all inputs are null (or if any input is null in some cases). Using COALESCE can help to substitute default values in such cases.
• Performance-wise, always prefer using Spark’s built-in SQL/functions over writing your own logic in Python (UDFs). The built-ins run within the Spark engine in a distributed manner and are highly optimized (often implemented in Java/Scala). We’ll discuss UDFs next, which are a way to extend functionality, but they come with a performance cost if misused.

More Spark tutorials

• about pyspark
• text diagram
• Apache Spark Runtime Architecture
• Introduction to RDD
• Actions vs Transformations
• Lazy Evaluation in PySpark

All tutorials · Try the free PySpark compiler · Practice challenges