XenorsPySpark DataFrames and Spark SQL for Beginners
Spark lets you query the same data with SQL or with the DataFrame API. Both compile to the same optimised plan, so choose whichever reads better for the task.
Create DataFrames with an explicit schema
Declaring a schema avoids a slow inference pass and prevents wrong types.
from pyspark.sql import SparkSession, functions as F
from pyspark.sql.types import StructType, StructField, StringType, IntegerType, DoubleType
spark = SparkSession.builder.master("local[*]").appName("sql").getOrCreate()
emp_schema = StructType([StructField("id", IntegerType()), StructField("name", StringType()),
StructField("dept_id", IntegerType()), StructField("salary", DoubleType())])
employees = spark.createDataFrame(
[(1, "Asha", 10, 72000.0), (2, "Ravi", 10, 65000.0), (3, "Meera", 20, 88000.0),
(4, "Kabir", 20, 91000.0), (5, "Zoya", 30, 54000.0)], emp_schema)
depts = spark.createDataFrame([(10, "Data"), (20, "Engineering"), (40, "Sales")], ["dept_id", "dept"])Select, filter and join
Join type matters: inner keeps matches only, left keeps all rows from the left side.
(employees.filter(F.col("salary") > 60000)
.select("name", (F.col("salary") / 12).alias("monthly"))
.show())
employees.join(depts, "dept_id", "left").show() # Zoya has no dept -> null
employees.join(depts, "dept_id", "inner").show()Write the same logic in Spark SQL
Register a temporary view and use ordinary SQL, including window functions.
employees.createOrReplaceTempView("employees")
depts.createOrReplaceTempView("depts")
spark.sql('''
SELECT d.dept, e.name, e.salary,
RANK() OVER (PARTITION BY d.dept ORDER BY e.salary DESC) AS rnk,
AVG(e.salary) OVER (PARTITION BY d.dept) AS dept_avg
FROM employees e JOIN depts d ON e.dept_id = d.dept_id
''').show()Read the execution plan
Use explain() to see whether Spark pushes filters down and which join strategy it picked. Small tables can be broadcast to every executor, which avoids an expensive shuffle.
from pyspark.sql.functions import broadcast
employees.join(broadcast(depts), "dept_id").explain()Save and reuse
Write results as Parquet, partitioned by a column you often filter on. Spark then reads only the partitions it needs.
employees.write.mode("overwrite").partitionBy("dept_id").parquet("out/employees")
spark.read.parquet("out/employees").filter("dept_id = 20").show()Hands-on practice
Run the examples in the downloadable Jupyter notebook. The extra checks below focus on mistakes that appear in real projects, not just syntax.
Verify a join with explicit keys
orders = spark.createDataFrame([(1,10),(2,20)], ['customer_id','amount'])
customers = spark.createDataFrame([(1,'A'),(3,'C')], ['customer_id','name'])
orders.join(customers,'customer_id','left').show()+-----------+------+----+ |customer_id|amount|name| +-----------+------+----+ |1 |10 |A | |2 |20 |NULL|
A left join preserves every order even when the customer dimension is incomplete.
Read the physical plan
q = orders.groupBy('customer_id').sum('amount')
q.explain(mode='formatted')Physical Plan +- HashAggregate ...
Use explain() before guessing about Spark performance.
Self-check
- Can you explain why the first practice result is trustworthy rather than just plausible?
- What would you monitor or test before using this Spark SQL & DataFrames workflow in production?
- Which assumption in the example would break first on a larger or messier dataset?
Frequently asked questions
DataFrame API or Spark SQL?
They are equivalent in performance. Use SQL when analysts will read it and the API when building reusable pipeline code.
What is a temporary view?
A name for a DataFrame that lets you query it with SQL inside the current Spark session.
What is a broadcast join?
Spark sends a small table to every executor so the large table does not need to be shuffled.
Primary references
Use these official sources when exact syntax, defaults or definitions matter beyond the examples here.
