PySpark – PySpark SQL
Table Of Contents:
- What is PySpark SQL?
- Why Use PySpark SQL?
- Setting It Up (Step-by-Step)
- SQL vs DataFrame APIs (Both Supported!)
- Advanced Features in PySpark SQL
- Input Data Formats
- Performance Optimizations
- Real-World Use Cases
- Summary
(1) What is PySpark SQL?
(2) Why Use PySpark SQL?
(3) Setting It Up (Step-by-Step)
Step 1: Create a SparkSession
from pyspark.sql import SparkSession
spark = SparkSession.builder \
.appName("PySparkSQLDemo") \
.getOrCreate()
Step 2: Load Data into a DataFrame
df = spark.read.csv("employees.csv", header=True, inferSchema=True)
df.show()
Step 3: Register DataFrame as SQL Table (Temp View)
df.createOrReplaceTempView("employees")
Step 4: Run SQL Queries!
result = spark.sql("""
SELECT department, AVG(salary) as avg_salary
FROM employees
GROUP BY department
""")
result.show()
(4) SQL vs DataFrame APIs (Both Supported!)
# SQL
spark.sql("SELECT * FROM employees WHERE age > 30").show()
# DataFrame API
df.filter(df.age > 30).show()
- You can mix and match both styles in your application.
(5) Advanced Features in PySpark SQL
Create SparkSession
from pyspark.sql import SparkSession
spark = SparkSession.builder.appName("App").getOrCreate()
Load Data
df = spark.read.csv("data.csv", header=True, inferSchema=True)
df = spark.read.json("data.json")
df = spark.read.parquet("data.parquet")
Basic DataFrame Operations
df.show() # Display top 20 rows
df.printSchema() # Print schema
df.columns # List of column names
df.describe().show() # Summary statistics
Filtering and Selection:
df.select("name", "age").show()
df.filter(df.age > 30).show()
df.where(df.salary > 50000).show()
Aggregations:
from pyspark.sql.functions import avg, max, min, count
df.groupBy("department").agg(avg("salary")).show()
df.groupBy("department").count().show()
With SQL
df.createOrReplaceTempView("employees")
spark.sql("SELECT * FROM employees WHERE age > 30").show()
spark.sql("""
SELECT department, AVG(salary)
FROM employees
GROUP BY department
""").show()
Join DataFrames
df1.join(df2, df1.id == df2.id, "inner").show()
Sorting & Limiting
df.orderBy("age").show()
df.sort(df.age.desc()).show()
df.limit(5).show()
Handling Missing Values
df.dropna().show() # Drop rows with nulls
df.fillna({"salary": 0, "name": "Unknown"}).show() # Fill nulls
Adding / Renaming Columns
df.withColumn("new_salary", df.salary * 1.1).show()
df.withColumnRenamed("old_name", "new_name")
Save Data
df.write.csv("output.csv", header=True)
df.write.json("output.json")
df.write.parquet("output.parquet")
(6) Input Data Formats
df = spark.read.parquet("data/employees.parquet")
df.createOrReplaceTempView("parquet_table")
spark.sql("SELECT * FROM parquet_table").show()
(7) Performance Optimizations
(8) Real-World Use Cases
(9) Summary
