SparkSession is the main entry point for working with Spark DataFrames, SQL, and all Spark functionality. It replaces the older SparkContext and SQLContext.

from pyspark.sql import SparkSession

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

RDD is a distributed, fault-tolerant, and immutable collection of objects that can be processed in parallel across a Spark cluster.

rdd = spark.sparkContext.parallelize([1, 2, 3, 4, 5])
print(rdd.collect())  # [1, 2, 3, 4, 5]

Key concepts:

• RDD is distributed across partitions
• RDDs are immutable — transformations create new RDDs
• Actions trigger execution
• Fault tolerance through lineage

data = [
    (1, "Praveen"),
    (2, "Ajay"),
    (3, "James")
]
columns = ["id", "name"]

df = spark.createDataFrame(data, columns)
df.show()
df.printSchema()

Output:

+---+-----+
| id| name|
+---+-----+
|  1|Praveen|
|  2| Ajay|
|  3|James|
+---+-----+

Schema:

root
 |-- id: long
 |-- name: string

df = spark.read.csv(
    "/content/data.csv",
    header=True,
    inferSchema=True
)
df.show()
df.printSchema()

header=True — The first row is treated as column names. Without it, Spark generates _c0, _c1, etc.

inferSchema=True — Spark attempts to determine appropriate data types from the data.

from pyspark.sql.types import (
    StructType, StructField, IntegerType, StringType
)

customer_schema = StructType([
    StructField("customer_id", IntegerType(), True),
    StructField("customer_name", StringType(), True),
    StructField("city", StringType(), True),
    StructField("age", IntegerType(), True),
    StructField("salary", IntegerType(), True)
])

df = spark.createDataFrame(data, customer_schema)

Why use an explicit schema?

• Better data quality
• Predictable data types
• Avoid incorrect schema inference
• Better production reliability

newdf = df.filter(df["Duration"] > 60)
newdf.show()

# Alternative using col()
from pyspark.sql.functions import col
newdf = df.filter(col("Duration") > 60)

Pandas vs PySpark:

# Pandas
df[df["age"] > 18]

# PySpark
df.filter(col("age") > 18)

from pyspark.sql.functions import col

df = df.withColumn(
    "difference_in_pulse",
    col("Maxpulse") - col("Pulse")
)
df.show()

from pyspark.sql.functions import avg

result = (
    employee_df
    .groupBy("department")
    .agg(avg("salary").alias("avg_salary"))
)
result.show()

Output:

+----------+---------+
|department|avg_salary|
+----------+---------+
|HR        |50000.0  |
|Engineering|65000.0|
+----------+---------+

# Ascending
df.orderBy(col("age").asc())

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

# Remove complete duplicate rows
new_df = df.dropDuplicates()

# Remove duplicates based on specific column
new_df = df.dropDuplicates(["customerid"])

Example: Orders and Customers

merged_df = orders_df.join(
    customer_df,
    on="customerid",
    how="inner"
)

new_df = merged_df.select("customerid", "amount", "age")
new_df.show()

Pandas equivalent:

pd.merge(df1, df2, on="customerid", how="inner")

Window functions are extremely important for PySpark interviews. Common functions: row_number(), rank(), dense_rank(), lag(), lead().

Second Highest Salary Per Department

Step 1 — Create Window Specification:

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

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

Step 2 — Apply dense_rank():

ranked_df = employee_df.withColumn(
    "rank",
    dense_rank().over(window_spec)
)

Step 3 — Filter Rank = 2:

second_highest = ranked_df.filter(col("rank") == 2)
second_highest.show()

A DataFrame can be registered as a temporary SQL view, then queried using SQL.

employee_df.createOrReplaceTempView("employee")

spark.sql("""
    SELECT * FROM employee
    WHERE department = 'HR'
""").show()

spark.sql("""
    SELECT department, AVG(salary) AS avg_salary
    FROM employee
    GROUP BY department
""").show()

Second Highest Salary Using SQL

spark.sql("""
    SELECT department, employee_name, salary
    FROM (
        SELECT department, employee_name, salary,
            DENSE_RANK() OVER (
                PARTITION BY department
                ORDER BY salary DESC
            ) AS rank
        FROM employee
    )
    WHERE rank = 2
""").show()

A UDF allows you to apply custom Python logic to Spark columns.

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

def to_uppercase(name):
    return name.upper()

uppercase_udf = udf(to_uppercase, StringType())

new_df = df.withColumn("uppercase_name", uppercase_udf(col("name")))

UDF vs Built-in Functions

Prefer built-in Spark functions whenever possible instead of Python UDFs because Python UDFs can introduce serialization and execution overhead.

# Instead of:
profit_udf(col("revenue"), col("expenses"))

# Prefer:
col("revenue") - col("expenses")

# Fill nulls with default values
df = df.fillna({"age": 30, "salary": 50000})

# Drop rows containing nulls
df = df.dropna()

# Count nulls by column
from pyspark.sql.functions import col, sum

null_counts = df.select([
    sum(col(c).isNull().cast("int")).alias(c)
    for c in df.columns
])
null_counts.show()

from pyspark.sql.functions import when, col

df = df.withColumn(
    "salary_category",
    when(col("salary") < 50000, "Low")
    .when((col("salary") >= 50000) & (col("salary") < 60000), "Medium")
    .otherwise("High")
)

Logic: salary < 50000 → Low, 50000–60000 → Medium, 60000+ → High

# Cache (MEMORY_ONLY by default)
df.cache()
df.filter(col("age") > 30).show()
df.groupBy("city").count().show()

# Persist with specific storage level
from pyspark import StorageLevel
df.persist(StorageLevel.MEMORY_AND_DISK)

# Remove cached data
df.unpersist()

# Repartition — full shuffle, can increase or decrease
df = df.repartition(20)

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

Simple rule: repartition → increase/decrease + shuffle, coalesce → reduce + less shuffle

from pyspark.sql.functions import broadcast

result = orders_df.join(
    broadcast(products_df),
    "product_id",
    "inner"
)

When to use: When one side of the join is small enough to fit in executor memory. This avoids shuffling the large dataset.

Pipeline: Source → Read → Transform → Validate → Write

from pyspark.sql import SparkSession
from pyspark.sql.functions import lit

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

# Read
df = spark.read.csv("s3://bucket/raw/data.csv", header=True, inferSchema=True)

# Transform
df = df.dropDuplicates()
df = df.withColumn("city", lit("Chennai"))

# Write
df.write.mode("overwrite").parquet("s3://bucket/transformed/")

jdbc_url = "jdbc:mysql://localhost:3306/database"
properties = {
    "user": "root",
    "password": "password",
    "driver": "com.mysql.cj.jdbc.Driver"
}

employee_df = spark.read.jdbc(
    url=jdbc_url,
    table="employee",
    properties=properties
)

department_df = spark.read.jdbc(
    url=jdbc_url,
    table="department",
    properties=properties
)

merged_df = employee_df.join(
    department_df,
    employee_df.departmentid == department_df.departmentid,
    "inner"
)

source = "s3://my-bucket/raw/data.csv"
destination = "s3://my-bucket/transformed/"

df = spark.read.csv(source, header=True, inferSchema=True)
new_df = df.dropDuplicates()
new_df.write.mode("overwrite").parquet(destination)
spark.stop()

Detect skew:

from pyspark.sql.functions import desc

df.groupBy("key") \
  .count() \
  .orderBy(desc("count")) \
  .show()

Salting technique: Add a random prefix to skewed keys to distribute data evenly.

Original: key=1 → 1,000,000 records

After salting:
1_0 → Partition 1
1_1 → Partition 2
1_2 → Partition 3
1_3 → Partition 4

Accumulators allow executors to contribute values that can be read by the driver. Useful for counting bad records during processing.

bad_records = spark.sparkContext.longAccumulator("bad_records")

def validate(row):
    if row.age < 18:
        bad_records.add(1)

print(bad_records.value)

Catalyst is Spark SQL's query optimization framework.

Spark Code / SQL
       ↓
Parsed Logical Plan
       ↓
Analyzed Logical Plan
       ↓
Optimized Logical Plan
       ↓
Physical Plan
       ↓
Execution

Key optimizations:

• Predicate Pushdown — Apply filters as early as possible
• Column Pruning — Read only required columns

AQE allows Spark to adapt parts of query execution using runtime information.

Initial Plan
     ↓
Execute
     ↓
Runtime Statistics
     ↓
AQE
     ↓
Adjusted Execution

Key features: coalescing shuffle partitions, changing join strategies, handling skewed joins.

# Basic plan
df.explain()

# Detailed plan with all stages
df.explain(True)

Use explain() to inspect whether Spark is using BroadcastHashJoin, SortMergeJoin, Exchange/shuffle, filtering, or projection.

Basic

• Create SparkSession
• Create RDD
• Create DataFrame
• Read CSV / JSON / Parquet
• Create manual schema

Transformations

• filter, select, withColumn
• groupBy, orderBy, join
• dropDuplicates, unionByName

Functions

• col, lit, when
• sum, avg, max, min, count

Window Functions

• row_number, rank, dense_rank
• lag, lead, partitionBy

Advanced

• UDF, Broadcast Join
• Cache, Persist, Repartition
• Data Skew, Salting, AQE
• explain(), Accumulators

Data Engineering

• JDBC, S3, ETL pipeline
• Data validation, Parquet output
• SCD Type 2, Incremental loading

For a hypothetical workload of around 5 GB, walk through the configuration process:

Input data
   ↓
Estimate partitions
   ↓
Estimate parallelism
   ↓
Choose executor cores
   ↓
Choose executor memory
   ↓
Run
   ↓
Inspect Spark UI
   ↓
Tune

A common starting-point calculation for partitions:

number of partitions ≈ data size / target partition size

But the appropriate partition size depends on: file format, compression, transformation, shuffle, cluster size, workload, and Spark configuration.

A typical spark-submit command with common configuration flags:

spark-submit \
  --num-executors 10 \
  --executor-cores 4 \
  --executor-memory 10g \
  --driver-memory 6g \
  myfile.py

Flag explanations:

• --num-executors — Number of executor processes.
• --executor-cores — Number of CPU cores available per executor.
• --executor-memory — Memory available to each executor.
• --driver-memory — Memory available to the driver.