Skip to content

Apache Spark Essentials

Apache Spark is a distributed computing engine for processing large datasets. This sheet covers SparkSession, DataFrames, transformations, actions, SQL, joins, aggregations, partitions, caching, and optimization.

Table of Contents

Installation / Setup

# Download Spark (or use pip for PySpark)
pip install pyspark

# SparkSession is the entry point
from pyspark.sql import SparkSession

spark = SparkSession.builder \
    .appName("MyApp") \
    .master("local[*]") \
    .getOrCreate()

# Access SparkContext
sc = spark.sparkContext

Core Syntax

SparkSession

# Create SparkSession
spark = SparkSession.builder \
    .appName("AppName") \
    .config("spark.sql.shuffle.partitions", "200") \
    .getOrCreate()

# Stop SparkSession
spark.stop()

Creating DataFrames

# From a list of tuples
data = [("Alice", 25), ("Bob", 30), ("Charlie", 35)]
df = spark.createDataFrame(data, ["name", "age"])

# From a list of dictionaries
data = [{"name": "Alice", "age": 25}, {"name": "Bob", "age": 30}]
df = spark.createDataFrame(data)

# From a Pandas DataFrame
import pandas as pd
pdf = pd.DataFrame({"name": ["Alice", "Bob"], "age": [25, 30]})
df = spark.createDataFrame(pdf)

# From a file
df = spark.read.csv("data.csv", header=True, inferSchema=True)
df = spark.read.parquet("data.parquet")
df = spark.read.json("data.json")
df = spark.read.orc("data.orc")

Transformations (lazy)

# Select columns
df.select("name", "age")

# Filter rows
df.filter(df.age > 25)
df.where("age > 25")

# Add/modify columns
from pyspark.sql.functions import col, lit
df.withColumn("age_next_year", col("age") + 1)
df.withColumn("country", lit("US"))

# Rename columns
df.withColumnRenamed("name", "first_name")

# Drop columns
df.drop("age")

# Sort
df.orderBy("age")
df.orderBy(col("age").desc())

# Distinct
df.distinct()

Actions (trigger computation)

# Show results
df.show()
df.show(5, truncate=False)

# Count rows
df.count()

# Collect to driver
rows = df.collect()
names = df.select("name").collect()

# First row
first = df.first()

# To Pandas
pdf = df.toPandas()

# Write output
df.write.parquet("output.parquet")
df.write.csv("output.csv", header=True)
df.write.mode("overwrite").parquet("output.parquet")

Common Operations

Aggregations

from pyspark.sql.functions import sum, avg, count, min, max, collect_list

# Single aggregation
df.groupBy("department").agg(
    avg("salary").alias("avg_salary"),
    count("*").alias("count")
)

# Multiple aggregations
df.groupBy("department", "level").agg(
    sum("salary").alias("total_salary"),
    count("distinct employee_id").alias("num_employees")
)

Joins

# Inner join
df1.join(df2, df1.id == df2.id, "inner")

# Left join
df1.join(df2, "id", "left")
# or with condition
df1.join(df2, df1.id == df2.id, "left")

# Right join
df1.join(df2, "id", "right")

# Full outer join
df1.join(df2, "id", "outer")

# Cross join
df1.crossJoin(df2)

# Semi join (like inner but only left columns)
df1.join(df2, "id", "left_semi")

# Anti join (left rows with no match)
df1.join(df2, "id", "left_anti")

Window Functions

from pyspark.sql.window import Window
from pyspark.sql.functions import row_number, rank, dense_rank, lead, lag

window = Window.partitionBy("department").orderBy("salary")

df.withColumn("rank", rank().over(window))
df.withColumn("row_num", row_number().over(window))
df.withColumn("prev_salary", lag("salary", 1).over(window))
df.withColumn("next_salary", lead("salary", 1).over(window))

SQL Queries

# Register DataFrame as temp view
df.createOrReplaceTempView("employees")

# Run SQL
result = spark.sql("""
    SELECT department, AVG(salary) as avg_salary
    FROM employees
    WHERE age > 25
    GROUP BY department
    HAVING COUNT(*) > 1
    ORDER BY avg_salary DESC
""")

Partitions

# Check partitions
df.rdd.getNumPartitions()

# Repartition (full shuffle)
df.repartition(10)
df.repartition("department")

# Coalesce (reduce partitions, no shuffle)
df.coalesce(5)

# Partition by column when writing
df.write.partitionBy("year", "month").parquet("output/")

Caching

# Cache in memory
from pyspark import StorageLevel
df.cache()  # or df.persist()

# Cache with storage level
df.persist(StorageLevel.MEMORY_AND_DISK)

# Unpersist
df.unpersist()

UDFs (User Defined Functions)

from pyspark.sql.functions import udf
from pyspark.sql.types import StringType

@udf(returnType=StringType())
def uppercase(name):
    return name.upper()

df.withColumn("name_upper", uppercase("name"))

Practical Examples

Example 1 — Data processing pipeline

from pyspark.sql import SparkSession
from pyspark.sql.functions import col, when, avg, count

spark = SparkSession.builder.appName("Pipeline").getOrCreate()

# Read data
df = spark.read.csv("sales.csv", header=True, inferSchema=True)

# Transform
result = df \
    .filter(col("amount") > 0) \
    .withColumn("category",
        when(col("amount") > 1000, "large")
        .when(col("amount") > 100, "medium")
        .otherwise("small")
    ) \
    .groupBy("region", "category") \
    .agg(
        count("*").alias("num_sales"),
        avg("amount").alias("avg_amount")
    ) \
    .orderBy("region", "category")

result.show()

Example 2 — Join and aggregate

# Load datasets
customers = spark.read.parquet("customers.parquet")
orders = spark.read.parquet("orders.parquet")

# Join and aggregate
customer_orders = customers.join(orders, "customer_id") \
    .groupBy("customer_id", "name") \
    .agg(
        count("order_id").alias("total_orders"),
        sum("amount").alias("total_spent")
    ) \
    .filter(col("total_spent") > 1000) \
    .orderBy(col("total_spent").desc())

customer_orders.show()

Example 3 — Window function for ranking

from pyspark.sql.window import Window
from pyspark.sql.functions import rank, dense_rank

window = Window.partitionBy("department").orderBy(col("salary").desc())

ranked = df.withColumn("salary_rank", rank().over(window)) \
    .withColumn("salary_dense_rank", dense_rank().over(window))

# Top 3 earners per department
ranked.filter(col("salary_rank") <= 3).show()

Common Patterns

Pattern: deduplication

# Remove duplicates based on specific columns
df.dropDuplicates(["id"])

# Remove exact duplicates
df.dropDuplicates()

Pattern: handling nulls

from pyspark.sql.functions import coalesce, when, col

# Drop rows with any nulls
df.dropna()

# Drop rows where specific column is null
df.dropna(subset=["name", "email"])

# Fill nulls
df.fillna(0)  # fill all numeric nulls
df.fillna({"name": "Unknown", "age": 0})

# Conditional null handling
df.withColumn("age", when(col("age").isNull(), 0).otherwise(col("age")))

Pattern: lazy evaluation

# Transformations are lazy — nothing happens until an action
df2 = df.filter(col("age") > 25)  # no computation yet
df3 = df2.select("name", "age")   # still no computation
result = df3.collect()             # triggers computation

# Check execution plan
df3.explain()

Best Practices

  • Use DataFrames over RDDs — DataFrames are optimized by Catalyst optimizer.
  • Avoid collect() on large datasets — use show(), take(), or write to storage.
  • Cache wisely — cache only when you reuse a DataFrame multiple times.
  • Partition for parallelism — repartition() for increasing, coalesce() for decreasing.
  • Use broadcast joins for small tables — from pyspark.sql.functions import broadcast.
  • Filter early — reduce data volume before expensive operations.
  • Use Parquet — columnar, compressed, schema-preserving.

Common Mistakes

Mistake Fix
Using collect() on large data Use show(), take(), or write output
Not caching reused DataFrames Call .cache() before multiple actions
Too many partitions Aim for 128MB–256MB per partition
Using Python UDFs over native functions Prefer built-in pyspark.sql.functions
Filtering after aggregations Filter before groupBy when possible
Ignoring data skew Use salting or broadcast joins

Troubleshooting

Symptom Cause / Fix
OutOfMemoryError Reduce partition size; increase executor memory
Slow performance Check for data skew; repartition; cache
Job failed with NPE Null handling — check for null values
Shuffle overflow Increase spark.sql.shuffle.partitions
Serialization error Use mapPartitions for heavy libraries

Official Documentation

Last Updated

2026-09-06 — Initial version.