🚀 PySpark Architecture & Execution Engine — Complete Guide

🔥 1. Spark Evolution Recap

• 2009: AMPLab (UC Berkeley)
• Goal: Overcome Hadoop MapReduce slowness
• 2013: Donated to Apache Foundation
• Built using Scala, runs on JVM, supports Python (via PySpark)

⚔️ 2. Spark vs Hadoop (Core Comparison)

🧱 3. Spark Design Goals

• Speed: 100x faster than Hadoop (in-memory)
• Unified Engine: Batch, Streaming, MLlib, GraphX
• Language APIs: Python (PySpark), Scala, Java, SQL
• Fault-Tolerant: Lineage tracking, RDDs
• Scalability: Tested with 8000+ nodes

🐍 4. Why PySpark? (Python + Spark Bridge)

PySpark is the Python API to the Spark engine, which is natively written in Scala/JVM.

🔄 How Python Code Reaches JVM:

[Python Code]
    ↓ py4j Gateway
[Python Driver Process]
    ↔
[JVM Driver (SparkContext)]
    ↓
[Executors (JVM Workers)]

🔧 What is py4j?

• A Java library embedded inside Spark that exposes JVM objects to Python.
• Python calls Java methods as if they were native — SparkContext, RDDs, etc.

✅ One-liner: PySpark uses py4j to route Python commands to the JVM-based Spark engine.

⚙️ 5. Master-Slave Architecture — Driver & Executors

🧠 Core Components:

🔁 Diagram: Simplified Flow

Python Code
   ↓
SparkSession → SparkContext
   ↓
DAG Scheduler → Stage Plan
   ↓
Task Scheduler → Tasks
   ↓
Executors (on Worker Nodes)
   ↓
Reads/Writes Data → Returns Result

🔁 6. DAG Scheduler & Task Lifecycle

🔹 What Happens When You Run .show()?

1. Logical Plan — based on DataFrame transformations
2. Optimized Plan — Catalyst optimizes for pushdowns, projections
3. Physical Plan — translates to stages
4. Stages — narrow (e.g., map) vs wide (e.g., join)
5. Tasks — 1 per partition, assigned to executors

✅ Lazy Evaluation: Transformations build a plan; actions trigger execution

🔌 7. py4j + JVM Bridge Deep Dive

🔧 JVM Roles

• PySpark driver runs in Python
• Real execution happens on the JVM via py4j
• Each RDD/DataFrame operation is wrapped and sent over

🔍 py4j Functions

🧵 8. Cluster Managers: Modes of Deployment

Examples:

spark-submit --master yarn --deploy-mode cluster script.py
spark-submit --master k8s://host --conf spark.kubernetes...

🗃️ 9. Spark Memory Hierarchy

• Storage Memory: For caching RDDs/DataFrames
• Execution Memory: Used while shuffles/sorts/joins happen
• Unified Memory Model: Shared space in executor
• Broadcast Memory: For small datasets to distribute across nodes

✅ Tune using:

--executor-memory 4g
--driver-memory 2g

📦 10. Monitoring via Spark UI

Access via:

• http://localhost:4040
• Or Spark History Server (for cluster jobs)

💾 11. Delta Lake + Databricks Essentials

Delta Lake runs on top of Spark + Parquet and adds:

Example:

df.write.format("delta").save("/mnt/delta_table")

# Upsert
from delta.tables import DeltaTable
DeltaTable.forPath(spark, "/mnt/table").alias("t") \
.merge(df.alias("s"), "t.id = s.id") \
.whenMatchedUpdateAll() \
.whenNotMatchedInsertAll() \
.execute()

🧪 12. Complete Execution Flow (Real Job)

# PySpark Job
df = spark.read.csv("hdfs://cluster/sales.csv")
df_filtered = df.filter("region = 'Asia'")
df_grouped = df_filtered.groupBy("product").sum("amount")
df_grouped.write.parquet("hdfs://results")

Internal Execution:

1. SparkSession → SparkContext initialized
2. DAG created (3 transformations)
3. Catalyst optimizes it
4. DAG → Stages (filter, groupBy)
5. Task Scheduler assigns tasks
6. YARN allocates executors
7. Executors run tasks per partition
8. Shuffle for groupBy
9. Result written to HDFS

🧾 13. Interview Cheat Sheet

✅ Terminologies to Memorize:

🧠 One-Line Summary

PySpark wraps Python logic and routes it to Spark’s JVM backend using py4j. The Driver constructs the job, builds a DAG of transformations, breaks it into stages, and assigns tasks to Executors across the cluster, which execute tasks in parallel, read/write from HDFS, and return results via the SparkContext — all monitored using the Spark UI.

Interview-Ready Cheat Sheet

Core Concepts

• RDD is the base abstraction → fault-tolerant, distributed
• DataFrame is a distributed table with schema
• Wide Transformations (e.g., join) cause shuffle
• Actions trigger execution (count(), show())

PySpark Interview One-Liners

• SparkSession is the unified entry point in PySpark ≥ 2.0
• Broadcast joins are optimal for small lookup tables
• Executor memory is not the same as driver memory
• Shuffle is expensive → avoid unless necessary
• Persist() vs Cache() → Cache = memory only, Persist = custom storage levels

🧠 One-Liner Architecture Summary

Apache Spark uses a Driver-Executor model, where the Driver acts as the master to coordinate work, Executors run distributed tasks, and a Cluster Manager allocates resources. Spark builds a DAG of transformations, splits it into stages and tasks, and executes them in parallel on executors, using memory/disk/shuffle/broadcast optimizations. Storage can be HDFS, S3, Delta Lake, or JDBC, with monitoring via the Spark UI and event logs.

🔚 Bonus: Real Job Flow Trace (Condensed for Interview)

1. Spark code is submitted → Driver starts
2. SparkContext connects to cluster
3. Transformations → DAG → Optimized Plan
4. Actions trigger DAG Scheduler
5. Cluster Manager allocates Executors
6. Tasks are assigned and run
7. Data read/shuffled/written by Executors
8. Results sent to Driver or written to output
9. Progress tracked in Web UI or History Server

Absolutely, Rajeev. Here’s your complete Spark Architecture Mind Map (in text format) and a mock interview round focused entirely on PySpark Architecture & Internals — deep, realistic, and industry-proven.

🧠 Mind Map: Spark Architecture (Text Outline Style)

Apache Spark Architecture
│
├── 1. High-Level Abstractions
│   ├── RDD: Resilient Distributed Dataset
│   ├── DataFrame: Schema-aware RDD
│   └── Dataset (Scala/Java only)
│
├── 2. Language Bindings
│   ├── Scala (Native)
│   ├── Java
│   ├── Python (PySpark via py4j)
│   └── R
│
├── 3. SparkSession
│   └── Entry point for DataFrame/SQL API
│
├── 4. Execution Components
│   ├── Driver
│   │   ├── Runs user code
│   │   ├── Hosts SparkContext
│   │   └── Builds DAG, manages job lifecycle
│   ├── Executors
│   │   ├── JVM processes on worker nodes
│   │   ├── Execute tasks
│   │   └── Cache data & shuffle blocks
│   └── Cluster Manager
│       ├── Standalone
│       ├── YARN
│       └── Kubernetes
│
├── 5. Internal Engines
│   ├── Catalyst Optimizer (logical/physical plan)
│   └── Tungsten (memory management & codegen)
│
├── 6. Execution Flow
│   ├── Transformations (lazy)
│   ├── Actions (trigger execution)
│   ├── Logical Plan → Optimized Plan → Physical Plan
│   ├── DAG Scheduler → Stages
│   └── Task Scheduler → Tasks → Executors
│
├── 7. Memory Management
│   ├── Execution Memory (shuffles, joins)
│   ├── Storage Memory (caching)
│   └── Reserved Memory
│
├── 8. Storage & IO
│   ├── HDFS / S3 / Azure / GCS
│   ├── JDBC / Hive Metastore
│   └── Delta Lake
│
├── 9. Monitoring Tools
│   ├── Spark UI (4040)
│   ├── History Server
│   └── Event Logs
│
├── 10. PySpark Internals
│   ├── Python API via py4j
│   ├── Python → JVM bridging
│   └── Serialization overhead considerations
│
└── 11. Optimizations
    ├── Broadcast joins
    ├── Caching / Persistence
    ├── Repartitioning / Coalesce
    └── AQE (Adaptive Query Execution)

You can visualize this as a circular or radial mind map with Apache Spark in the center.

🎯 Mock Interview: PySpark Architecture & Internals

Level: Mid to Senior
Format: Q&A + Suggested Ideal Answers

Q1. What is the role of the Driver in Spark?

✅ Answer:
The Driver is the master process that runs the user’s main function and coordinates Spark jobs. It contains the SparkContext, builds the DAG, converts it into stages and tasks, and sends tasks to executors.

Q2. What happens when you call an action like .count() on a DataFrame?

✅ Answer:

• A logical plan is built based on prior transformations
• Catalyst optimizes it into an optimized and physical plan
• DAG Scheduler splits it into stages
• Task Scheduler assigns tasks to executors
• Results are aggregated and returned to the driver

Q3. How does PySpark communicate with the Spark engine?

✅ Answer:
PySpark uses a library called py4j to send Python commands to the JVM backend, which executes them using Spark’s Scala-based engine. The results are passed back via py4j.

Q4. What is the difference between an Executor and a Task?

✅ Answer:
An Executor is a JVM process launched on worker nodes responsible for executing code and storing data.
A Task is the smallest unit of work sent to an executor, and corresponds to one data partition.

Q5. What are narrow and wide transformations?

✅ Answer:

• Narrow transformations (e.g., map, filter) do not require shuffling — data stays on the same node.
• Wide transformations (e.g., groupBy, join) require shuffle, moving data across the network.

Q6. How is memory managed in Spark Executors?

✅ Answer:
Memory in an executor is split into:

• Execution memory: Used for operations like joins/sorts.
• Storage memory: For caching RDD/DataFrame.
• Reserved memory: For internal Spark needs.
  Unified Memory Model (default since Spark 1.6) shares space between execution and storage.

Q7. What is Catalyst Optimizer?

✅ Answer:
Catalyst is Spark’s query optimization engine for transforming logical plans into efficient physical plans. It supports constant folding, predicate pushdown, projection pruning, etc.

Q8. How do you handle small lookup tables during joins?

✅ Answer:
Use Broadcast Join — small DataFrame is broadcast to all executors to avoid shuffle.

from pyspark.sql.functions import broadcast
df.join(broadcast(df_small), "key").show()

Q9. What is the role of py4j in PySpark?

✅ Answer:
py4j is the gateway that allows Python code to invoke JVM methods. It bridges the Python driver with Spark’s Scala backend.

Q10. Explain the complete life of a Spark job.

✅ Answer:

1. SparkSession starts; SparkContext initialized
2. Lazy transformations build a logical plan
3. Catalyst optimizes it → physical plan
4. DAG scheduler breaks it into stages
5. Task scheduler schedules tasks per partition
6. Executors run tasks and return results
7. Spark UI shows status of all components

Bonus: ⚠️ Common Pitfalls

Absolutely, Rajeev! You’re in elite mastery mode now. Let’s hit all three topics in depth:

🧱 DELTA LAKE ARCHITECTURE & TIME TRAVEL — INTERNALS

✅ What is Delta Lake?

Delta Lake is a storage layer on top of Parquet that brings ACID transactions, schema enforcement, time travel, and merge operations to Apache Spark.

🧰 Delta Lake Key Features

🔬 Internals: How Delta Lake Works

1. Each write creates a new version (JSON + Parquet files).
2. Transaction log is stored in _delta_log directory.
3. Read operations refer to the latest version (or past if versionAsOf is used).
4. All changes are append-only — no overwrite of existing data files.
5. Delta maintains a commit log (00000000.json) + checkpointing for fast access.

📦 Delta Time Travel

# Read latest
df = spark.read.format("delta").load("/mnt/sales")

# Read version 5
df_v5 = spark.read.format("delta").option("versionAsOf", 5).load("/mnt/sales")

# Read as of timestamp
df_time = spark.read.format("delta").option("timestampAsOf", "2024-12-01 12:00:00").load("/mnt/sales")

🧪 Merge (Upsert) Example

from delta.tables import DeltaTable

delta_table = DeltaTable.forPath(spark, "/mnt/sales")

delta_table.alias("t").merge(
    updates_df.alias("u"),
    "t.id = u.id"
).whenMatchedUpdateAll().whenNotMatchedInsertAll().execute()

🧹 Optimize & Vacuum

spark.sql("OPTIMIZE delta.`/mnt/sales` ZORDER BY (region, product)")

spark.sql("VACUUM delta.`/mnt/sales` RETAIN 168 HOURS")

📁 RAW → BRONZE → SILVER → GOLD: LAKEHOUSE PIPELINE DESIGN

🔶 1. Raw Layer (Landing)

• Source: CSVs, JSON, Kafka, APIs, RDBMS
• Storage: No schema enforcement, audit only
• Examples: /datalake/raw/sales/

raw_df = spark.read.option("multiline", True).json("/raw/sales/")

🟫 2. Bronze Layer (Cleaned & Ingested)

• Validated, minimally cleaned
• Add metadata: _ingest_date, _source
• Write to Delta Table
• Partitioned by key like date or region

bronze_df = raw_df.withColumn("_ingest_date", current_date())
bronze_df.write.format("delta").mode("overwrite").save("/bronze/sales/")

⚪ 3. Silver Layer (Validated + Transformed)

• Business-ready format
• Joins with lookup tables
• Conform dimensions (e.g., DateDim, ProductDim)
• Column renaming, derived metrics

silver_df = bronze_df \
    .filter("amount > 0") \
    .join(region_dim, "region_id") \
    .withColumn("profit", col("amount") - col("cost"))

silver_df.write.format("delta").mode("overwrite").save("/silver/sales/")

🟡 4. Gold Layer (Aggregated/Reporting)

• Dashboard-ready format
• KPI aggregations, rollups
• Materialized views

gold_df = silver_df.groupBy("region", "month").agg(
    sum("profit").alias("monthly_profit"),
    count("*").alias("txn_count")
)

gold_df.write.format("delta").mode("overwrite").save("/gold/sales_summary/")

💡 BONUS: Real-Time Trigger (Auto Loader + Streaming)

raw_stream = spark.readStream.format("cloudFiles") \
    .option("cloudFiles.format", "json") \
    .load("/raw/sales")

bronze_stream = raw_stream.withColumn("_ingest_time", current_timestamp())

bronze_stream.writeStream.format("delta") \
    .option("checkpointLocation", "/checkpoints/bronze/") \
    .start("/bronze/sales")

📋 MOCK INTERVIEW ROUND 2 — Delta Lake, Lakehouse, and Pipelines

Q1. How does Delta Lake achieve ACID transactions in Spark?

✅ Answer:
Delta uses a transaction log (_delta_log) that records every commit as a JSON file, similar to a WAL (write-ahead log). Operations are atomic, and concurrent writes are serialized.

Q2. What is the difference between Delta Lake and regular Parquet files?

✅ Answer:

Q3. How would you implement SCD Type 2 using Delta Merge?

✅ Answer:
Use merge() with whenMatchedUpdate, whenNotMatchedInsert, and maintain a current_flag, start_date, end_date column to track changes over time.

Q4. What is the purpose of Bronze/Silver/Gold architecture?

✅ Answer:

• Bronze: Raw cleaned data, audit trails
• Silver: Business logic applied, conform dimensions
• Gold: Aggregated, report-ready data
  It enforces data quality, version control, and modularity.

Q5. How is Time Travel internally maintained in Delta Lake?

✅ Answer:
Each write creates a new snapshot/version in _delta_log. The log stores metadata and the exact set of files for that version. versionAsOf or timestampAsOf fetches that version during read.

Q6. How do you prevent VACUUM from deleting old data prematurely?

✅ Answer:
Delta has a default 7-day retention (168 hours). Set lower manually only if absolutely safe:

SET spark.databricks.delta.retentionDurationCheck.enabled = false;
VACUUM ... RETAIN 1 HOURS

Q7. How do you ensure schema consistency in Bronze → Silver flow?

✅ Answer:
Use mergeSchema=True when writing OR set strict enforcement to avoid corrupting downstream pipelines.

df.write.option("mergeSchema", True).format("delta")...

Pages: 1 2 3