The DataFrame API and Spark SQL

What we're doing

You'll learn the DataFrame API and Spark SQL side by side by writing the same queries both ways.

Step 1: DataFrames recap

In the RDDs and DataFrames tutorial you learned that a DataFrame is a distributed table, basically rows and columns with a defined schema, split across the cluster. Spark stores them in a columnar format, applies the Catalyst optimizer to rewrite your queries for performance, and executes lazily until you call an action.

Two ways to use it:

• DataFrame API — Python method chains like df.filter(...).select(...)
• Spark SQL — actual SQL strings like "SELECT ... FROM ... WHERE ..." passed to spark.sql()

Step 2: Start PySpark

pyspark --master spark://localhost:7077

You'll get a Python prompt with spark set up as your cluster connection.

Step 3: Load both DataFrames

airlines = spark.read.csv("/home/preparesh/spark-data/airlines.csv", header=True, inferSchema=True)
flights = spark.read.csv("/home/preparesh/spark-data/flights.csv", header=True, inferSchema=True)

airlines.show()
flights.show()

Two DataFrames loaded and printed as tables.

Step 4: The DataFrame API — select, filter, withColumn

Basic query operations using the DataFrame API:

Select specific columns:

flights.select("flight_id", "origin", "destination").show()

Filter rows:

flights.filter(flights.delay_minutes > 60).show()

Combine select and filter:

flights.filter(flights.delay_minutes > 60).select("flight_id", "origin", "destination", "delay_minutes").show()

Add a new column:

from pyspark.sql.functions import lit

flights.withColumn("data_source", lit("realtime_feed")).show()

withColumn adds a new column to the DataFrame. lit("realtime_feed") means literally the string "realtime_feed" for every row. lit is Spark's way of saying "this isn't a column reference, it's a constant value."

Step 5: Joins with the DataFrame API

joined = flights.join(airlines, on="airline_code")
joined.show()

Join flights with airlines on the airline_code column. Now each flight row also has the airline's name and country attached to it.

You can control this with a how parameter:

flights.join(airlines, on="airline_code", how="left").show()

Find the average delay per airline:

from pyspark.sql.functions import avg

joined.groupBy("name").agg(avg("delay_minutes").alias("avg_delay")).show()

Step 6: Spark/actual SQL

Now the same operations but written as SQL strings.

First, register the DataFrames as SQL tables:

airlines.createOrReplaceTempView("airlines")
flights.createOrReplaceTempView("flights")

createOrReplaceTempView registers the DataFrame under a name so SQL queries can find it. Temp means it only exists for this Spark session.

Now you can query them with actual SQL:

spark.sql("SELECT flight_id, origin, destination FROM flights WHERE delay_minutes > 60").show()

Spark parses the string, converts it to the same execution plan as the DataFrame API, and runs it.

The join and aggregation from before, in SQL:

spark.sql("""
    SELECT a.name, AVG(f.delay_minutes) AS avg_delay
    FROM flights f
    JOIN airlines a ON f.airline_code = a.airline_code
    GROUP BY a.name
""").show()

Exit the session with exit().