Apache Spark

Overview

Apache Spark is a unified analytics engine for large-scale data processing. Originally developed at UC Berkeley’s AMPLab in 2009, Spark was designed to overcome the limitations of Hadoop MapReduce by enabling in-memory computing, which can be up to 100x faster for iterative workloads. Spark was open-sourced in 2010 and became an Apache top-level project in 2014.

Key capabilities include:

• In-memory computing: Keeps intermediate data in RAM, dramatically speeding up iterative algorithms and interactive queries
• Unified platform: Single engine for batch processing, streaming, SQL, machine learning, and graph computation
• Multi-language APIs: Native support for Scala, Java, Python (PySpark), R, and SQL
• Fault tolerance: Automatic recovery through lineage-based recomputation of lost partitions
• Lazy evaluation: Builds an optimized execution plan (DAG) before executing, enabling whole-stage code generation

Architecture & Core Components

Cluster Architecture

Spark follows a master-worker architecture:

┌─────────────────────────────────────────────────┐
│                  Driver Program                  │
│  ┌─────────────┐  ┌──────────┐  ┌────────────┐  │
│  │ SparkContext │  │ DAG      │  │ Task       │  │
│  │             │  │ Scheduler│  │ Scheduler  │  │
│  └─────────────┘  └──────────┘  └────────────┘  │
└──────────────────────┬──────────────────────────┘
                       │
          ┌────────────┼────────────┐
          ▼            ▼            ▼
   ┌─────────────┐ ┌─────────────┐ ┌─────────────┐
   │  Executor 1 │ │  Executor 2 │ │  Executor N │
   │ ┌────┬────┐ │ │ ┌────┬────┐ │ │ ┌────┬────┐ │
   │ │Task│Task│ │ │ │Task│Task│ │ │ │Task│Task│ │
   │ └────┴────┘ │ │ └────┴────┘ │ │ └────┴────┘ │
   │   Cache     │ │   Cache     │ │   Cache     │
   └─────────────┘ └─────────────┘ └─────────────┘

Core Components

1. Driver Program

• Runs the main() function and creates the SparkContext
• Converts user program into a DAG (Directed Acyclic Graph) of stages
• Negotiates resources with the Cluster Manager
• Schedules tasks on executors and monitors progress

2. Cluster Manager

• Allocates resources across applications
• Supported managers: Standalone, YARN, Mesos, Kubernetes
• Manages executor lifecycle (launch, monitor, restart)

3. Executors

• JVM processes running on worker nodes
• Execute tasks and store data in memory or disk
• Report task status and results back to the driver
• Each executor has a fixed number of cores and memory

4. DAG Scheduler

• Translates RDD transformations into a DAG of stages
• Narrow dependencies (map, filter): pipelined within a stage
• Wide dependencies (groupByKey, join): require a shuffle boundary between stages
• Optimizes the execution plan by merging narrow transformations

5. Task Scheduler

• Assigns tasks to executors based on data locality
• Locality levels: PROCESS_LOCAL → NODE_LOCAL → RACK_LOCAL → ANY
• Handles task failures with configurable retry logic

Execution Flow

1. User submits a Spark application (driver program)
2. Driver creates SparkContext and connects to Cluster Manager
3. Cluster Manager allocates executors on worker nodes
4. Driver sends application code (JAR/Python files) to executors
5. SparkContext sends tasks to executors for execution
6. Executors run tasks, cache data, and return results to driver

Resilient Distributed Datasets (RDDs)

RDDs are the fundamental abstraction in Spark — an immutable, distributed collection of objects that can be processed in parallel.

Key Properties

• Immutable: Once created, cannot be modified (transformations create new RDDs)
• Distributed: Partitioned across cluster nodes
• Resilient: Can be recomputed from lineage if a partition is lost
• Lazy: Transformations are not executed until an action is called

Operations

Transformations (lazy, return new RDD):

• map(func) — Apply function to each element
• filter(func) — Select elements where function returns true
• flatMap(func) — Map then flatten
• groupByKey() — Group values by key
• reduceByKey(func) — Merge values for each key
• join(otherRDD) — Inner join on keys
• coalesce(n) / repartition(n) — Change partition count

Actions (trigger execution, return results):

• collect() — Return all elements to driver
• count() — Count elements
• reduce(func) — Aggregate elements
• saveAsTextFile(path) — Write to storage
• foreach(func) — Apply function to each element (side effects)

Persistence / Caching

rdd.persist(StorageLevel.MEMORY_AND_DISK)

Storage levels:

Use unpersist() to remove from cache. Spark also automatically evicts old partitions using LRU.

DataFrames and Datasets

DataFrames

• Distributed collection of data organized into named columns (like a relational table)
• Schema-aware — enables the Catalyst optimizer to optimize queries
• Available in all languages (Python, Scala, Java, R)

# Create DataFrame
df = spark.read.json("people.json")

# Query with DataFrame API
df.filter(df.age > 21) \
  .groupBy("department") \
  .agg({"salary": "avg"}) \
  .show()

Datasets (Scala/Java only)

• Type-safe version of DataFrames
• Compile-time type checking
• Encoder-based serialization (faster than Java serialization)

Catalyst Optimizer

Spark SQL’s query optimizer that applies rule-based and cost-based optimizations:

1. Analysis: Resolves column names and types using the catalog
2. Logical Optimization: Predicate pushdown, constant folding, column pruning
3. Physical Planning: Generates multiple physical plans, selects best via cost model
4. Code Generation: Whole-stage code generation compiles query plans into optimized Java bytecode

Spark SQL

Spark SQL provides a programming interface for structured data processing using SQL or the DataFrame API.

Key features:

• Hive compatibility: Read/write Hive tables, use HiveQL
• JDBC/ODBC: Connect BI tools (Tableau, Power BI) via Thrift server
• Schema inference: Automatically infer schema from JSON, Parquet, CSV
• Unified API: Mix SQL queries with DataFrame transformations

# Register DataFrame as SQL temp view
df.createOrReplaceTempView("employees")

# Run SQL query
result = spark.sql("""
    SELECT department, AVG(salary) as avg_salary
    FROM employees
    WHERE age > 25
    GROUP BY department
    HAVING AVG(salary) > 50000
    ORDER BY avg_salary DESC
""")

Supported Data Sources

• File formats: Parquet (default), ORC, JSON, CSV, Avro, Text
• Databases: JDBC (MySQL, PostgreSQL, Oracle), Hive, Cassandra
• Streaming: Kafka, Kinesis, files
• Cloud: S3, ADLS, GCS

Spark Streaming

Structured Streaming (recommended)

Built on the Spark SQL engine, treats streaming data as an unbounded table:

# Read from Kafka
df = spark.readStream \
    .format("kafka") \
    .option("kafka.bootstrap.servers", "host:9092") \
    .option("subscribe", "topic1") \
    .load()

# Process
result = df.selectExpr("CAST(value AS STRING)") \
    .groupBy("value") \
    .count()

# Write to console
query = result.writeStream \
    .outputMode("complete") \
    .format("console") \
    .start()

Output modes:

• Append — Only new rows (for non-aggregation queries)
• Complete — Entire result table (for aggregations)
• Update — Only changed rows

Triggers:

• Default (micro-batch as fast as possible)
• Fixed interval (processingTime="10 seconds")
• Once (single micro-batch, then stop)
• Continuous (experimental, ~1ms latency)

DStreams (legacy)

• Original streaming API based on micro-batches of RDDs
• Still supported but Structured Streaming is preferred

Watermarking

Handles late-arriving data in streaming:

df.withWatermark("eventTime", "10 minutes") \
  .groupBy(window("eventTime", "5 minutes")) \
  .count()

Spark MLlib

Machine learning library providing:

• Classification: Logistic Regression, Decision Trees, Random Forest, GBT, SVM
• Regression: Linear, Generalized Linear, Decision Tree, Random Forest
• Clustering: K-means, Gaussian Mixture, LDA, Bisecting K-means
• Recommendation: ALS (Alternating Least Squares)
• Feature engineering: TF-IDF, Word2Vec, StandardScaler, PCA, VectorAssembler
• ML Pipelines: Chain transformers and estimators for reproducible workflows

from pyspark.ml.pipeline import Pipeline
from pyspark.ml.classification import LogisticRegression
from pyspark.ml.feature import VectorAssembler, StandardScaler

assembler = VectorAssembler(inputCols=["age", "salary"], outputCol="features")
scaler = StandardScaler(inputCol="features", outputCol="scaled")
lr = LogisticRegression(featuresCol="scaled", labelCol="label")

pipeline = Pipeline(stages=[assembler, scaler, lr])
model = pipeline.fit(trainingData)
predictions = model.transform(testData)

GraphX

Graph computation library providing:

• Graph construction: From edge lists, vertex/edge RDDs
• Graph algorithms: PageRank, Connected Components, Triangle Counting, Shortest Paths
• Pregel API: Vertex-centric iterative computation (like Google’s Pregel)
• Integration with Spark SQL via GraphFrames (separate package)

Deployment Modes

Client Mode (default)

• Driver runs on the machine that submitted the application
• Good for interactive use (notebooks, spark-shell)
• Driver must stay alive for the application’s duration

Cluster Mode

• Driver runs on a worker node inside the cluster
• Good for production jobs submitted remotely
• Application continues even if submitting machine disconnects

Cluster Manager Comparison

Performance Tuning

Memory Management

• Execution memory: Used for shuffles, joins, sorts, aggregations
• Storage memory: Used for caching/persisting RDDs
• Unified memory: Execution and storage share a region (default since Spark 1.6), either can borrow from the other

Key configs:

• spark.executor.memory — Total executor heap (default 1g)
• spark.memory.fraction — Fraction of heap for execution + storage (default 0.6)
• spark.memory.storageFraction — Storage’s share within that fraction (default 0.5)

Shuffle Optimization

• Shuffles are the most expensive operation (disk I/O + network)
• Reduce shuffles: use reduceByKey instead of groupByKey
• spark.sql.shuffle.partitions — Number of partitions after shuffle (default 200)
• Broadcast joins: Broadcast small tables to avoid shuffle

from pyspark.sql.functions import broadcast
result = large_df.join(broadcast(small_df), "key")

Data Skew

• Salting: Add random prefix to skewed keys, join, then remove prefix
• Adaptive Query Execution (AQE): Spark 3.0+ automatically handles skew

Serialization

• Kryo serializer: 10x faster than Java serialization
• spark.serializer = org.apache.spark.serializer.KryoSerializer

Partitioning

• Aim for 2-4 partitions per CPU core
• Partition size: 128MB–200MB is optimal
• repartition(n) for increasing partitions (full shuffle)
• coalesce(n) for decreasing partitions (no shuffle)

Common Pitfalls

1. Collecting large datasets to driver: collect() on a large RDD causes OOM on the driver
2. Using groupByKey: Shuffles all data; use reduceByKey or aggregateByKey instead
3. Not caching reused RDDs: Recomputes from scratch each time
4. Too few / too many partitions: Too few = underutilization, too many = scheduling overhead
5. Serialization of non-serializable objects: Closures that reference outer objects cause NotSerializableException
6. Ignoring data locality: Moving data across the network is 100x slower than local reads

Interview Questions

Conceptual

1. What is the difference between RDD, DataFrame, and Dataset?

   • RDD: Low-level, unstructured, no optimization. DataFrame: Schema-aware, optimized via Catalyst. Dataset: Type-safe DataFrame (Scala/Java only).

2. What are narrow vs wide transformations?

   • Narrow: Each parent partition maps to at most one child partition (map, filter). Wide: Each parent partition maps to multiple child partitions requiring a shuffle (groupByKey, join).

3. How does Spark achieve fault tolerance?

   • Through RDD lineage. If a partition is lost, Spark recomputes it from the parent RDDs using the recorded transformations (DAG).

4. What is lazy evaluation and why does Spark use it?

   • Transformations are not executed immediately — they build a DAG. This allows the Catalyst optimizer to rearrange and optimize the entire pipeline before execution.

5. Explain the Catalyst optimizer.

   • Multi-phase optimizer: analysis → logical optimization → physical planning → code generation. Applies predicate pushdown, constant folding, join reordering, and whole-stage code generation.

Architecture

6. What happens when you submit a Spark job?

   • Driver creates SparkContext → connects to Cluster Manager → requests executors → sends code to executors → DAG Scheduler splits job into stages → Task Scheduler assigns tasks → executors run tasks and return results.

7. How does Spark handle data skew?

   • Techniques include salting keys, broadcast joins for small tables, repartitioning, and Adaptive Query Execution (AQE) in Spark 3.0+.

8. When would you use persist() vs cache()?

   • cache() is shorthand for persist(MEMORY_AND_DISK). Use persist() when you need a specific storage level (e.g., MEMORY_ONLY_SER for memory-constrained environments).

Performance

9. How would you optimize a slow Spark job?

   • Check the Spark UI for stage durations and shuffle sizes. Common fixes: broadcast small tables, use reduceByKey over groupByKey, tune partition count, enable AQE, cache frequently accessed data, use columnar formats (Parquet).

10. What is the difference between repartition and coalesce?

    • repartition(n) performs a full shuffle to create exactly n partitions (can increase or decrease). coalesce(n) avoids a full shuffle by merging existing partitions (can only decrease). Use coalesce when reducing partitions for better performance.

Spark vs Hadoop MapReduce