---
title: "Spark SQL interpreter PySpark"
canonical: "https://onesaitplatform.refined.site/space/DOCT/4982931468/Spark%20SQL%20interpreter%20PySpark"
format: markdown
---
> Macro (toc)

# Introduction

This tutorial will explain how to use **Spark SQL** in PySpark (based on Spark 2.x).

[https://spark.apache.org/sql/](https://spark.apache.org/sql/)

Before we start, a couple of Spark SQL concepts need to be clarified:

- **SparkSession**: this is the entry point of Spark SQL, and you need to use SparkSession to create a DataFrame/Dataset, register UDFs, query tables and etc.
- **DataFrame**: there are no Datasets as such in PySpark, only DataFrames. The PySpark DataFrame is very similar to the Pandas DataFrame concept, but it is distributed.

There are two ways to create a **DataFrame**:

- Using SparkSession to create the DataFrame directly: from RDD, List objects, etc.
- Using DataFrameReader to create a DataFrame from many storage types supported by Spark, such as HDFS, JDBC, etc.

That said, here are some examples of how to work with SparkSQL.

> ℹ️ It is recommended to first run a paragraph with ‘%spark.conf’ to make sure that the correct configuration is used.

# Example of use

## Creating a Dataset/DataFrame via SparkSession

In the following example a Spark DataFrame will be created from any data. In addition to displaying it as a table, the data schema will be displayed.

The same can be done using Pandas.

```
%pyspark

# create DataFrame from python list. It can infer schema for you.
df1 = spark.createDataFrame([(1, "andy", 20, "USA"), (2, "jeff", 23, "China"), (3, "james", 18, "USA")]).toDF("id", "name", "age", "country")
df1.printSchema()
df1.show()

# create DataFrame from pandas dataframe
df2 = spark.createDataFrame(df1.toPandas())
df2.printSchema()
df2.show()
```

Executing the code, the result will be as follows:

![image-20250213-103010.png](media://3ee85672-43ee-4335-bfba-c4a4e66f89c8)


## Creating a DataFrame using DataFrameReader

In the following example, data will be read from different files; a JSON and a CSV. The DataFrameReader will be used for this.

```
%pyspark

SPARK_HOME = os.getenv('SPARK_HOME')

# Read data from json file
# link for this people.json (https://github.com/apache/spark/blob/master/examples/src/main/resources/people.json)
# Use hdfs path if you are using hdfs
df1 = spark.read.json("file://" + SPARK_HOME + "/examples/src/main/resources/people.json")
df1.printSchema()
df1.show()

# Read data from csv file. You can customize it via spark.read.options. E.g. In the following example, we customize the sep and header
df2 = spark.read.options(sep=";", header=True).csv("file://"  + SPARK_HOME + "/examples/src/main/resources/people.csv")
df2.printSchema()
df2.show()

# Specify schema for your csv file
from pyspark.sql.types import StructType, StringType, IntegerType

schema = StructType().add("name", StringType(), True) \
    .add("age", IntegerType(), True) \
    .add("job", StringType(), True)
    
df3 = spark.read.options(sep=";", header=True) \
    .schema(schema) \
    .csv("file://" + SPARK_HOME + "/examples/src/main/resources/people.csv") 
```

By executing the paragraph, the result will look like this:

![image-20250213-111247.png](media://73cfafdc-da54-4fa4-ba07-ad48f0fb6c84)

## Adding a new column

Even if you generate a table in the DataFrame, or import the table from a file, they are not locked and are open to add more data to it. With the following code, you generate a table and then add new columns to it.

```
%pyspark

# withColumn could be used to add new Column
df1 = spark.createDataFrame([(1, "andy", 20, "USA"), (2, "jeff", 23, "China"), (3, "james", 18, "USA")]).toDF("id", "name", "age", "country")

df2 = df1.withColumn("age2", df1["age"] + 1)
df2.show()

# the new column could replace the existing the column if the new column name is the same as the old column
df3 = df1.withColumn("age", df1["age"] + 1)
df3.show()

# Besides using expression to create new column, you could also use udf to create new column
# Use F.upper instead of upper, because the builtin udf of spark may conclifct with that of python, such as max
import pyspark.sql.functions as F
df4 = df1.withColumn("name", F.upper(df1["name"]))
df4.show()
```

The result of executing the paragraph is as follows:

![image-20250213-111701.png](media://62f8793d-a351-4b03-b46f-f9e54d7ee0d8)

## Delete a column

Similarly, starting from the name of a column header, you can delete it without much trouble with the following code:

```
%pyspark

df1 = spark.createDataFrame([(1, "andy", 20, "USA"), (2, "jeff", 23, "China"), (3, "james", 18, "USA")]).toDF("id", "name", "age", "country")
# drop could be used to remove Column
df2 = df1.drop("id")
df2.show()
```

Based on the table in the previous example, the result would look like this:

![image-20250213-112340.png](media://18404663-cbe4-43ab-a5ec-517d56088391)

## Select subset of columns

Continuing with the columns, it is possible to select a subset of the columns. This can be done as shown below:

```
%pyspark

df1 = spark.createDataFrame([(1, "andy", 20, "USA"), (2, "jeff", 23, "China"), (3, "james", 18, "USA")]).toDF("id", "name", "age", "country")
# select can accept a list of string of the column names
df2 = df1.select("id", "name")
df2.show()

# select can also accept a list of Column. You can create column via $ or udf
import pyspark.sql.functions as F
df3 = df1.select(df1["id"], F.upper(df1["name"]), df1["age"] + 1)
df3.show()
```

The result would be as follows:

![image-20250213-112649.png](media://df744974-0884-4207-a874-a869560f1814)

## Column filtering

It is also possible to filter the content of columns according to simple filters, both numeric and text. This can be seen in the following code:

```
%pyspark

df1 = spark.createDataFrame([(1, "andy", 20, "USA"), (2, "jeff", 23, "China"), (3, "james", 18, "USA")]).toDF("id", "name", "age", "country")

# filter accept a Column 
df2 = df1.filter(df1["age"] >= 20)
df2.show()

# To be noticed, you need to use "&" instead of "&&" or "AND" 
df3 = df1.filter((df1["age"] >= 20) & (df1["country"] == "China"))
df3.show()
```

By executing the code, the table generated with the DataFrame is filtered according to the age or country fields:

![image-20250213-130923.png](media://b6418a7e-8a6c-423f-bb85-9dd09c3e5547)

## User-defined functions

There are certain times when it is necessary to interact with data in a more elaborate way than mere filters. For this, it is possible to use Python Spark functions (**PySpark UDF**).

[https://sparkbyexamples.com/pyspark/pyspark-udf-user-defined-function/](https://sparkbyexamples.com/pyspark/pyspark-udf-user-defined-function/)

The following example shows how to use a Python lambda together with a custom function:

```
%pyspark

df1 = spark.createDataFrame([(1, "andy", 20, "USA"), (2, "jeff", 23, "China"), (3, "james", 18, "USA")]) \
            .toDF("id", "name", "age", "country")

# Create udf create python lambda
from pyspark.sql.functions import udf
udf1 = udf(lambda e: e.upper())
df2 = df1.select(udf1(df1["name"]))
df2.show()

# UDF could also be used in filter, in this case the return type must be Boolean
# We can also use annotation to create udf
from pyspark.sql.types import *
@udf(returnType=BooleanType())
def udf2(e):
    if e >= 20:
        return True;
    else:
        return False

df3 = df1.filter(udf2(df1["age"]))
df3.show()

# UDF could also accept more than 1 argument.
udf3 = udf(lambda e1, e2: e1 + "_" + e2)
df4 = df1.select(udf3(df1["name"], df1["country"]).alias("name_country"))
df4.show()
```

By launching the paragraph, the result would look like this:

![image-20250213-140726.png](media://dc0beb6b-233b-4e06-8b45-95dc51c259fd)

## Groupings

As with SQL, it is possible to perform groupings according to the parameters defined in the paragraph code. An example would be the following code which presents several types of groupings:

- Group the data by country and counting the number of records for each.
- Group the data by country, calculating the average age and counting the number of records for each country (without and with aliases).
- Group the data by country, calculating the average age and maximum age for each country.

```
%pyspark

df1 = spark.createDataFrame([(1, "andy", 20, "USA"), (2, "jeff", 23, "China"), (3, "james", 18, "USA")]) \
           .toDF("id", "name", "age", "country")

# You can call agg function after groupBy directly, such as count/min/max/avg/sum
df2 = df1.groupBy("country").count()
df2.show()

# Pass a Map if you want to do multiple aggregation
df3 = df1.groupBy("country").agg({"age": "avg", "id": "count"})
df3.show()

import pyspark.sql.functions as F
# Or you can pass a list of agg function
df4 = df1.groupBy("country").agg(F.avg(df1["age"]).alias("avg_age"), F.count(df1["id"]).alias("count"))
df4.show()

# You can not pass Map if you want to do multiple aggregation on the same column as the key of Map should be unique. So in this case
# you have to pass a list of agg functions
df5 = df1.groupBy("country").agg(F.avg(df1["age"]).alias("avg_age"), F.max(df1["age"]).alias("max_age"))
df5.show()
```

Tras ejecutar el párrafo, el resultado sería:

![image-20250213-142158.png](media://5366ab6a-ce8b-4e6f-b327-f3c81560fd6b)

## Simple joins

Another type of operation that can be performed is the simple join from a single field. Thus, given two DataFrames, it will be indicated from which field to make the join.

```
%pyspark

df1 = spark.createDataFrame([(1, "andy", 20, 1), (2, "jeff", 23, 2), (3, "james", 18, 3)]).toDF("id", "name", "age", "c_id")
df1.show()

df2 = spark.createDataFrame([(1, "USA"), (2, "China")]).toDF("c_id", "c_name")
df2.show()

# You can just specify the key name if join on the same key
df3 = df1.join(df2, "c_id")
df3.show()

# Or you can specify the join condition expclitly in case the key is different between tables
df4 = df1.join(df2, df1["c_id"] == df2["c_id"])
df4.show()

# You can specify the join type afte the join condition, by default it is inner join
df5 = df1.join(df2, df1["c_id"] == df2["c_id"], "left_outer")
df5.show()
```

The result of executing the paragraph would show the following links:

![image-20250213-142516.png](media://97f21427-24ab-49ec-9302-ac4b8cf7877b)

## Complex connections

It will also be possible to make more complex joins from multiple fields. In the following example you will create two DataFrames, one with personal information and one with country information, and then make a join between the two datasets using two common fields to combine the data.

```
%pyspark

df1 = spark.createDataFrame([("andy", 20, 1, 1), ("jeff", 23, 1, 2), ("james", 12, 2, 2)]).toDF("name", "age", "key_1", "key_2")
df1.show()

df2 = spark.createDataFrame([(1, 1, "USA"), (2, 2, "China")]).toDF("key_1", "key_2", "country")
df2.show()

# Join on 2 fields: key_1, key_2

# You can pass a list of field name if the join field names are the same in both tables
df3 = df1.join(df2, ["key_1", "key_2"])
df3.show()

# Or you can specify the join condition expclitly in case when the join fields name is differetnt in the two tables
df4 = df1.join(df2, (df1["key_1"] == df2["key_1"]) & (df1["key_2"] == df2["key_2"]))
df4.show()
```

After executing the paragraph, the following result will be displayed:

![image-20250213-143506.png](media://1ea25117-b50f-4e7b-8e8a-b72f6e8de43f)

## Using SQL directly

It will also be possible to launch SQL queries. As an example, in the following paragraph a DataFrame will be created with people data, registering in a temporary view to perform SQL queries, to select columns and to apply a custom function (UDF).

```
%pyspark

df1 = spark.createDataFrame([(1, "andy", 20, "USA"), (2, "jeff", 23, "China"), (3, "james", 18, "USA")]) \
           .toDF("id", "name", "age", "country")
# call createOrReplaceTempView first if you want to query this DataFrame via sql
df1.createOrReplaceTempView("people")
# SparkSession.sql return DataFrame
df2 = spark.sql("select name, age from people")
df2.show()

# You need to register udf if you want to use it in sql
spark.udf.register("udf1", lambda e : e.upper())
df3 = spark.sql("select udf1(name), age from people")
df3.show()
```

After the execution, the following tables will be displayed:

![image-20250213-144018.png](media://b5058b97-dc88-4ad0-be4a-f72a5c96d87b)

## Display tables

Although data has already been displayed in tabular form, it is possible to represent the information in more advanced tables, with options and a graphical interface that is more pleasing to the eye.

With the following code you will be able to display the table:

```
%spark.sql

show tables
```

Executing the paragraph, you get this result:

![image-20250213-144807.png](media://84ddc244-0bdc-41c2-a5c0-8a18fc4e10a3)

## Display a DataFrame/Dataset using z.show

In addition to representing the data in tabular form, it can also be represented in different types of graphs. An example is the following, in which the data can be represented in the following way.

```
%pyspark

df1 = spark.createDataFrame([(1, "andy", 20, "USA"), (2, "jeff", 23, "China"), (3, "james", 18, "USA")]).toDF("id", "name", "age", "country")
df2 = df1.groupBy("country").count()
z.show(df2)
```

After launching the paragraph, the data would be displayed in graphical form:

![image-20250213-153619.png](media://5c558282-b2d2-4f92-bbcc-3cf24e96f912)

## Other types of visualisations

There are other ways to visualise the data, so we encourage you to try out the available options.

In the following example using spark.sql you can see a visualisation using a pie chart.

```
%spark.sql

select country, count(1) as count from people group by country
```

Executing the paragraph will display the following graph:

![image-20250213-153952.png](media://755c118e-3e90-47a9-ae42-7d9296ef8789)

# Conclusions

Thanks to PySpark, it is possible to represent and work with tabular data in a simple and agile way. It is therefore a very useful complement when working with Notebooks, allowing the data analyst to have a wide range of possibilities for both the analysis and the representation of data.

# Download example

The example used in this tutorial is available for download below: