DataFrames, SQL & Transformations

Reading Data & Basic Operations

from pyspark.sql import SparkSession
from pyspark.sql.functions import col, avg, concat, lit

spark = SparkSession.builder.appName('MyApp').getOrCreate()

# Parquet stores schema natively — inference is cheap and exact
df = spark.read.parquet('s3://bucket/orders.parquet')

# CSV/JSON require sampling to guess types — an explicit schema is
# faster and avoids wrong guesses on large/messy files
df_csv = spark.read.option('header', True).csv('data.csv')

df.printSchema()
df.show(5)  # ACTION — triggers computation, prints a preview

df.select('name', 'amount').filter(col('amount') > 100).show()
df.withColumn('full_name', concat(col('first'), lit(' '), col('last')))

Lazy Evaluation: Transformations vs. Actions

# Transformations (select, filter, groupBy, join, withColumn) are LAZY —
# they build a query plan, nothing runs yet.
result = (
    df.filter(col('status') == 'completed')
      .groupBy('region')
      .agg(avg('amount').alias('avg_amount'))
)

# Actions (show, collect, count, write) trigger actual execution —
# Catalyst optimizes the WHOLE chain above right before this runs
result.show()

SQL Queries & Joins

df.createOrReplaceTempView('orders')

# SQL and the DataFrame API compose freely — spark.sql() returns a DataFrame too
spark.sql('''
  SELECT region, AVG(amount) as avg_amount
  FROM orders
  WHERE status = 'completed'
  GROUP BY region
''').show()

orders.join(customers, orders.customer_id == customers.id, 'inner')
# join types: inner, left, right, outer, left_semi, left_anti

Window Functions

from pyspark.sql import Window
from pyspark.sql.functions import rank

# "Top N per group" — a rank across rows without collapsing them, unlike groupBy
window = Window.partitionBy('department').orderBy(col('salary').desc())
df.withColumn('rank', rank().over(window)).filter(col('rank') <= 3)