🔥 PySpark Cheat Sheet

Complete PySpark DataFrame API — reads, transforms, joins, aggregations, window functions, UDFs, and performance tuning.

SparkSession & Reading Data

SparkSession

from pyspark.sql import SparkSession

spark = SparkSession.builder \
    .appName('my_app') \
    .config('spark.sql.shuffle.partitions', 200) \
    .config('spark.executor.memory', '4g') \
    .getOrCreate()

spark.sparkContext.setLogLevel('WARN')

Initialize Spark with custom configuration.

Read Data

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')

# With options
df = spark.read.format('csv') \
    .option('header', True) \
    .option('delimiter', '|') \
    .option('nullValue', 'NA') \
    .schema(custom_schema) \
    .load('path/*.csv')

# JDBC
df = spark.read.jdbc(url, table, properties={'user':'u','password':'p'})

Read CSV, Parquet, JSON, ORC, Delta, and JDBC.

Define Schema

from pyspark.sql.types import (
    StructType, StructField,
    StringType, IntegerType, FloatType,
    DateType, TimestampType, ArrayType, MapType
)

schema = StructType([
    StructField('name', StringType(), nullable=False),
    StructField('age', IntegerType(), True),
    StructField('salary', FloatType(), True),
    StructField('tags', ArrayType(StringType()), True),
])

df = spark.read.schema(schema).csv('data.csv', header=True)

Explicit schemas for type safety and faster reads.

Create DataFrame Manually

data = [('Alice', 30), ('Bob', 25)]
df = spark.createDataFrame(data, ['name', 'age'])

# From Pandas
import pandas as pd
pdf = pd.DataFrame({'a': [1,2], 'b': [3,4]})
df = spark.createDataFrame(pdf)

Create DataFrames from lists or Pandas.

Inspection & Schema

Inspect DataFrame

df.printSchema()         # Column names & types
df.show(5, truncate=False)
df.display()              # Databricks notebooks
df.describe().show()      # Summary stats
df.summary().show()       # Extended stats
df.count()                # Total rows
df.columns                # List of column names
df.dtypes                 # [(name, type), ...]
df.schema                 # StructType object
df.isEmpty()              # True if empty

All ways to inspect DataFrame structure and content.

Select & Column Ops

from pyspark.sql.functions import col, lit, expr

df.select('name', 'age')
df.select(col('salary').alias('pay'))
df.select(expr('salary * 12 AS annual'))
df.selectExpr('name', 'salary * 12 AS annual')
df.withColumn('bonus', col('salary') * 0.1)
df.withColumn('country', lit('US'))
df.withColumnRenamed('old_name', 'new_name')
df.drop('temp_col')
df.toDF('new_col1', 'new_col2')  # Rename all

Select, rename, create, and drop columns.

Filtering & Sorting

Filter / Where

df.filter(col('age') > 30)
df.filter((col('dept') == 'Eng') & (col('salary') > 80000))
df.filter(col('name').isin('Alice', 'Bob'))
df.filter(col('email').isNotNull())
df.filter(col('name').like('A%'))
df.filter(col('name').rlike('^[A-M]'))  # Regex
df.filter(col('age').between(25, 35))
df.filter(~col('status').isin('inactive','banned'))  # NOT

All filtering operators — comparison, regex, IN, BETWEEN.

Sort & Limit

df.orderBy('salary')                    # Ascending
df.orderBy(col('salary').desc())        # Descending
df.orderBy('dept', col('salary').desc()) # Multi-column
df.sort(col('date').asc_nulls_last())   # NULLs last
df.limit(100)                           # Top N rows

Sort by columns and limit result count.

Distinct & Drop Duplicates

df.distinct()                     # Full row dedup
df.dropDuplicates(['user_id'])    # Dedup on columns
df.dropDuplicates(['user_id', 'date'])  # Composite dedup

Remove duplicate rows.

Joins & Unions

Joins

# Basic join
joined = df1.join(df2, on='id', how='inner')

# Types: inner, left, right, full, cross, semi, anti
df1.join(df2, on='id', how='left')
df1.join(df2, on='id', how='left_anti')  # NOT IN equivalent
df1.join(df2, on='id', how='left_semi')  # EXISTS equivalent
df1.crossJoin(df2)                       # Cartesian

# Multi-column join
df1.join(df2, on=['id', 'date'], how='left')

# Different column names
df1.join(df2, df1['user_id'] == df2['id'], 'left')

All join types including semi and anti joins.

Union & Intersect

# Union (same schema required)
df1.union(df2)             # Keep duplicates
df1.unionAll(df2)          # Same as union
df1.unionByName(df2, allowMissingColumns=True)

df1.intersect(df2)         # Common rows
df1.exceptAll(df2)         # Rows in df1 not in df2

Stack, intersect, or subtract DataFrames.

Aggregations

GroupBy

from pyspark.sql.functions import (
    count, sum, avg, max, min,
    countDistinct, collect_list, collect_set, stddev
)

df.groupBy('dept').agg(
    count('*').alias('total'),
    countDistinct('user_id').alias('unique_users'),
    sum('revenue').alias('total_rev'),
    avg('salary').alias('avg_salary'),
    max('salary').alias('max_salary'),
    min('salary').alias('min_salary'),
    stddev('salary').alias('std_salary'),
    collect_list('name').alias('names'),
    collect_set('role').alias('unique_roles')
)

Complete set of aggregation functions.

Pivot & Unpivot

# Pivot: rows to columns
df.groupBy('year').pivot('quarter', ['Q1','Q2','Q3','Q4']) \
  .agg(sum('revenue'))

# Unpivot / Melt: columns to rows
from pyspark.sql.functions import expr
df.selectExpr(
    'id',
    "stack(3, 'Q1',Q1, 'Q2',Q2, 'Q3',Q3) AS (quarter, revenue)"
).filter('revenue IS NOT NULL')

Reshape data between wide and long formats.

Rollup & Cube

# Rollup: hierarchical subtotals
df.rollup('region', 'dept').agg(sum('sales').alias('total'))

# Cube: all combination subtotals
df.cube('region', 'dept').agg(sum('sales').alias('total'))

Multi-level subtotals and cross-tabulations.

Window Functions

Window Setup

from pyspark.sql.window import Window
from pyspark.sql.functions import (
    row_number, rank, dense_rank, ntile,
    lag, lead, first, last,
    sum as _sum, avg as _avg
)

w = Window.partitionBy('dept').orderBy(col('salary').desc())
w_unbound = Window.partitionBy('dept').orderBy('date') \
    .rowsBetween(Window.unboundedPreceding, Window.currentRow)
w_range = Window.partitionBy('dept').orderBy('date') \
    .rangeBetween(-7, 0)  # 7 days lookback

Define window specs with partition, order, and frame.

Ranking

df.withColumn('rn', row_number().over(w))
df.withColumn('rnk', rank().over(w))
df.withColumn('drnk', dense_rank().over(w))
df.withColumn('quartile', ntile(4).over(w))
df.withColumn('pct', percent_rank().over(w))

All ranking functions — row_number, rank, dense_rank, ntile.

LAG, LEAD & Running Totals

df.withColumn('prev_val', lag('amount', 1).over(w))
df.withColumn('next_val', lead('amount', 1, 0).over(w))
df.withColumn('running_total', _sum('amount').over(w_unbound))
df.withColumn('moving_avg', _avg('amount').over(
    Window.partitionBy('store').orderBy('date')
    .rowsBetween(-6, 0)  # 7-day moving avg
))
df.withColumn('first_val', first('amount').over(w))
df.withColumn('last_val', last('amount').over(w))

Lag, lead, running totals, moving averages, first/last.

String & Date Functions

String Functions

from pyspark.sql.functions import (
    upper, lower, trim, ltrim, rtrim,
    length, substring, concat, concat_ws,
    split, regexp_replace, regexp_extract,
    when, coalesce, lpad, rpad, initcap
)

df.withColumn('upper_name', upper('name'))
df.withColumn('parts', split('full_name', ' '))
df.withColumn('clean', regexp_replace('text', '[^a-zA-Z]', ''))
df.withColumn('area', regexp_extract('phone', r'\((\d+)\)', 1))
df.withColumn('padded', lpad('id', 10, '0'))
df.withColumn('merged', concat_ws('-', 'year', 'month', 'day'))

All string manipulation functions.

Date & Timestamp

from pyspark.sql.functions import (
    current_date, current_timestamp,
    year, month, dayofmonth, dayofweek, hour,
    date_format, to_date, to_timestamp,
    datediff, months_between, date_add, date_sub,
    date_trunc, last_day, next_day
)

df.withColumn('yr', year('date_col'))
df.withColumn('mo', month('date_col'))
df.withColumn('fmt', date_format('ts', 'yyyy-MM-dd HH:mm'))
df.withColumn('parsed', to_date('str_col', 'MM/dd/yyyy'))
df.withColumn('days', datediff('end_date', 'start_date'))
df.withColumn('next_week', date_add('date_col', 7))
df.withColumn('month_start', date_trunc('month', 'date_col'))

Complete date/timestamp extraction and arithmetic.

Null Handling & Conditional

NULL Handling

df.filter(col('email').isNotNull())
df.filter(col('phone').isNull())
df.na.drop()                           # Drop rows with any null
df.na.drop(subset=['email', 'name'])   # Drop if these are null
df.na.fill(0)                          # Fill all numeric nulls
df.na.fill({'age': 0, 'name': 'Unknown'})
df.na.replace(['NA', 'N/A', ''], [None, None, None])

Filter, drop, and fill NULL values.

WHEN / OTHERWISE

from pyspark.sql.functions import when, coalesce

df.withColumn('level',
    when(col('salary') > 100000, 'Senior')
    .when(col('salary') > 60000, 'Mid')
    .otherwise('Junior')
)

df.withColumn('contact', coalesce('phone', 'email', lit('N/A')))

Conditional column logic and null coalescing.

UDFs & Spark SQL

User Defined Functions

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

@udf(returnType=StringType())
def classify(salary):
    if salary > 100000: return 'Senior'
    elif salary > 60000: return 'Mid'
    return 'Junior'

df.withColumn('level', classify('salary'))

# Pandas UDF (vectorized, much faster)
from pyspark.sql.functions import pandas_udf
import pandas as pd

@pandas_udf('double')
def normalize(s: pd.Series) -> pd.Series:
    return (s - s.mean()) / s.std()

df.withColumn('norm_salary', normalize('salary'))

Custom scalar UDFs and fast Pandas vectorized UDFs.

Spark SQL

df.createOrReplaceTempView('employees')

result = spark.sql("""
    SELECT dept, AVG(salary) AS avg_sal
    FROM employees
    WHERE status = 'active'
    GROUP BY dept
    HAVING AVG(salary) > 80000
""")

result.show()

Run SQL queries directly on DataFrames.

Writing Data

Write Output

df.write.parquet('output/data.parquet')
df.write.csv('output/data.csv', header=True)
df.write.json('output/data.json')

# With options
df.write.mode('overwrite') \
    .partitionBy('year', 'month') \
    .parquet('output/')

# Modes: overwrite, append, ignore, error
df.coalesce(1).write.mode('overwrite').csv('single_file/')

Save to various formats with partitioning.

Performance & Optimization

Partitioning & Caching

df.rdd.getNumPartitions()       # Check partition count
df.repartition(10, 'date')      # Hash repartition (shuffle)
df.coalesce(1)                  # Reduce partitions (no shuffle)

df.cache()                      # MEMORY_ONLY
df.persist(StorageLevel.MEMORY_AND_DISK)
df.unpersist()                  # Release

spark.catalog.clearCache()      # Clear all caches

Control partitioning and cache strategies.

Broadcast Join

from pyspark.sql.functions import broadcast

# Force broadcast for small table (< 10MB)
result = big_df.join(broadcast(small_df), on='key')

# Check broadcast threshold
spark.conf.get('spark.sql.autoBroadcastJoinThreshold')
spark.conf.set('spark.sql.autoBroadcastJoinThreshold', '50m')

Broadcast small tables to avoid shuffle in joins.

Explain & Debug

df.explain()          # Physical plan
df.explain(True)      # All plans (parsed, analyzed, optimized, physical)

# Check data skew
df.groupBy(spark_partition_id()).count().show()

# AQE (Adaptive Query Execution)
spark.conf.set('spark.sql.adaptive.enabled', True)
spark.conf.set('spark.sql.adaptive.coalescePartitions.enabled', True)

Analyze plans, detect skew, and tune execution.