PySpark Interview Questions & Answers

# RDD: No schema, opaque Java objects
rdd = sc.parallelize([('Alice', 30)])

# DataFrame: Named columns, Catalyst optimized
df = spark.createDataFrame([('Alice', 30)], ['name', 'age'])

df1 = spark.read.csv('data.csv') # Lazy (No execution)
df2 = df1.filter(col('age') > 21) # Lazy (No execution)
df3 = df2.select('name') # Lazy (No execution)
df3.count() # ACTION: Triggers DAG build & physical plan execution!

# Narrow (No Shuffle): filter, map, select, flatMap, union
df_narrow = df.filter(col('status') == 'ACTIVE')

# Wide (Triggers Network Shuffle): groupBy, join, distinct, repartition, reduceByKey
df_wide = df.groupBy('department').count()

# Shuffle triggered by Wide transformation:
df.groupBy('country').sum('revenue')

# Reduce 100 partitions to 10 efficiently without shuffle:
df_coalesced = df.coalesce(10)

# Increase 10 partitions to 50 or rebalance skewed data (Full Shuffle):
df_repartitioned = df.repartition(50)

rdd = sc.parallelize([('a', 1), ('a', 2), ('b', 3)])
# Efficient: Combines locally on mapper node before sending across network
rdd.reduceByKey(lambda x, y: x + y)

orders_df.join(broadcast(country_lookup_df), 'country_code')

# Add random salt (0-9) to skewed dataframe:
df_salted = df.withColumn('salted_key', concat(col('user_id'), lit('_'), floor(rand() * 10)))

# Driver OOM Fix: Avoid collect(), increase driver RAM:
--driver-memory 8g

# Executor OOM Fix: Increase executor RAM, fix data skew, lower shuffle partitions:
--executor-memory 16g --conf spark.memory.fraction=0.8

# AQE automatically applies 3 key runtime optimizations:
# 1. Coalesces small shuffle partitions dynamically
# 2. Converts Sort-Merge Join to Broadcast Join at runtime
# 3. Handles Skew Joins automatically

import pandas as pd
from pyspark.sql.functions import pandas_udf

@pandas_udf('double')
def vectorized_tax(salary: pd.Series) -> pd.Series:
    return salary * 0.2

df.withColumn('tax', vectorized_tax(col('salary')))

# For vanilla Parquet: Coalesce before write
df.coalesce(5).write.parquet('/path')

# For Delta Lake: Run OPTIMIZE and ZORDER
spark.sql('OPTIMIZE delta_table ZORDER BY (date, customer_id)')

# 4 Catalyst Stages:
# 1. Analysis (Catalog verification of column names/types)
# 2. Logical Optimization (Predicate pushdown, column pruning)
# 3. Physical Planning (Selecting join algorithms: Broadcast vs Sort-Merge)
# 4. Code Generation (WholeStageCodegen compiling Java bytecode)

from pyspark import StorageLevel

# cache() uses default MEMORY_AND_DISK in DataFrames:
df.cache()

# persist() allows custom storage levels:
df.persist(StorageLevel.MEMORY_ONLY_SER)

# Broadcast Variable (Read-only on workers):
b_lookup = sc.broadcast({'US': 'United States', 'CA': 'Canada'})

# Accumulator (Write-only aggregate from workers):
err_counter = sc.accumulator(0)

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

w = Window.partitionBy('dept').orderBy(col('salary').desc())
df.select('salary',
    row_number().over(w).alias('rn'), # 1, 2, 3
    rank().over(w).alias('rk'),       # 1, 1, 3 (skips 2 on tie)
    dense_rank().over(w).alias('dr')  # 1, 1, 2 (no gap on tie)
)

# Delta Lake Features:
# 1. ACID Transactions (Serializability)
# 2. Time Travel (Query past table versions)
# 3. MERGE INTO (Upserts & Deletes)
# 4. Schema Enforcement & Schema Evolution

# Joining Fact table partitioned by date_id with filtered Date Dim table:
fact_sales.join(dim_date.filter(col('year') == 2024), 'date_id')

df.withWatermark('event_time', '10 minutes')\
  .groupBy(window('event_time', '5 minutes'), 'device_id')\
  .count()

# Client Mode: Driver runs on submit machine (Notebooks/IDE)
spark-submit --deploy-mode client script.py

# Cluster Mode: Driver runs inside worker node in cluster (Production)
spark-submit --deploy-mode cluster script.py