SPARKS

Last reviewed: 2026-06-16

Purpose: Comprehensive Knowledge Base (KB) article on Apache Spark — covering architecture, core data abstractions, Spark SQL, PySpark DataFrames, performance tuning, Kubernetes deployment, and a complete ETL example.

---

Table of Contents

1. Overview
2. Spark Architecture
3. Core Data Abstractions: RDD vs DataFrame vs Dataset
4. Spark SQL
5. PySpark DataFrame Operations
6. Performance Tuning
7. Spark on Kubernetes
8. Complete PySpark ETL Example
9. References

---

Overview

Apache Spark is a unified, open-source, distributed computing engine for large-scale data processing. It provides high-level APIs in Java, Scala, Python (PySpark), and R, along with an optimized engine that supports general execution graphs. Key capabilities include:

• Batch and stream processing under a single unified API (Structured Streaming)
• SQL analytics via Spark SQL with the Catalyst optimizer
• Machine learning through MLlib
• Graph processing via GraphX

Spark runs on standalone clusters, Hadoop YARN, Apache Mesos, and Kubernetes. It can read from diverse data sources: HDFS, S3, GCS, ADLS, JDBC, Cassandra, Kafka, and more.

---

Spark Architecture

A Spark application follows a master–worker model with a single driver and multiple executors, coordinated by a cluster manager.

+-------------------------------------------------------+
|                     Driver Program                      |
|  SparkContext / SparkSession                           |
|  DAG Scheduler | Task Scheduler | Backend Scheduler    |
+--------------------------+----------------------------+
                           |
                    Cluster Manager
              (Standalone / YARN / K8s / Mesos)
                           |
          +----------------+------------------+
          |                                   |
     Executor 1                          Executor N
  +-------------+                     +-------------+
  | Cache/Tasks | ...                 | Cache/Tasks |
  +-------------+                     +-------------+

Driver

• Runs the user's main() function and creates a SparkContext (or SparkSession in modern Spark).
• Converts user code into a DAG (Directed Acyclic Graph) of stages.
• Schedules tasks across executors via the cluster manager.
• Holds metadata about all persisted RDDs/DataFrames.

Executors

• Worker processes that run tasks and store data in memory or disk.
• Start once the application begins and shut down when it ends.
• Report task progress and cached block status back to the driver.

Cluster Manager

Allocates resources across applications. Options:

Manager	Description	Typical Use Case
Standalone	Simple built-in cluster manager	Dev/test, small clusters
Hadoop YARN	Integrates with Hadoop ecosystem, resource-aware	Enterprise Hadoop deployments
Kubernetes	Container-native orchestration, elastic scaling	Cloud-native, microservices shops
Apache Mesos	General resource manager (Spark < 3.x)	Legacy multi-framework clusters

Execution Flow

1. Driver creates a SparkSession and connects to the cluster manager.
2. Cluster manager launches executors and allocates cores/memory.
3. Driver sends task code to executors.
4. Executors run tasks, cache data, and return results.
5. Driver collects results or writes them to external storage.

---

Core Data Abstractions: RDD vs DataFrame vs Dataset

Feature	RDD	DataFrame	Dataset (JVM only)
Introduced	Spark 1.0	Spark 1.3	Spark 1.6
Type safety	Yes (compile-time)	No (runtime)	Yes (compile-time)
Schema	No (raw Java/Python objects)	Yes (tabular, named columns)	Yes (strongly-typed JVM objects)
Optimization	Manual (no Catalyst)	Catalyst optimizer + Tungsten	Catalyst + Tungsten + encoders
Serialization	Java/Kryo (slow)	Tungsten binary (fast)	Tungsten binary (fast)
API	Java, Scala, Python, R	Java, Scala, Python, R	Java, Scala
Use when	Need low-level control, custom transformations	High-level SQL-like analytics, performance matters	Type-safe transformations on the JVM

When to use which

• RDD: Only when working with unstructured data (e.g., raw text streams) or when you need fine-grained control over partitioning and persistence.
• DataFrame: Default choice for structured/semi-structured data (JSON, Parquet, Avro, CSV). Leverages Catalyst for query optimization.
• Dataset: Use in Scala/Java when you need compile-time type safety on structured data (e.g., domain objects with case classes).

PySpark Note: Python does not support the typed Dataset API. In PySpark, DataFrame is equivalent to Dataset[Row].

---

Spark SQL

Spark SQL enables SQL queries on structured data, either via SQL strings or programmatic DataFrame operations.

Creating temporary views

from pyspark.sql import SparkSession

spark = SparkSession.builder \
    .appName("SparkSQLDemo") \
    .getOrCreate()

df = spark.read.json("s3://bucket/events/")
df.createOrReplaceTempView("events")

result = spark.sql("""
    SELECT user_id, COUNT(*) AS event_count
    FROM events
    WHERE event_type = 'purchase'
    GROUP BY user_id
    ORDER BY event_count DESC
    LIMIT 10
""")
result.show()

Catalyst Optimizer

Catalyst is Spark SQL's query optimizer. It performs:

1. Analysis — Resolves column names, types, and table references
2. Logical optimization — Predicate pushdown, constant folding, projection pruning
3. Physical planning — Chooses join strategies (broadcast vs sort-merge), bucketing
4. Code generation — Tungsten generates optimized JVM bytecode

You can view the query plan with:

df.explain("formatted")    # Physical plan
df.explain("cost")         # With cost estimates (Spark 3.x)

Reading and writing with Spark SQL

# Parquet — columnar, highly optimized
df = spark.read.parquet("hdfs:///data/events/")
df.write.mode("overwrite").parquet("hdfs:///data/events_clean/")

# Delta Lake — ACID transactions, time travel
df.write.format("delta").save("/data/delta/events/")

# JDBC — relational databases
df = spark.read \
    .format("jdbc") \
    .option("url", "jdbc:postgresql://host:5432/db") \
    .option("dbtable", "public.orders") \
    .option("user", "user") \
    .option("password", "pass") \
    .load()

---

PySpark DataFrame Operations

Below are the most common PySpark DataFrame operations with examples.

Setup

from pyspark.sql import SparkSession
from pyspark.sql.functions import col, lit, when, sum, avg, count, \
    upper, regexp_extract, to_date, date_format, current_timestamp
from pyspark.sql.types import StructType, StructField, StringType, IntegerType, DoubleType

spark = SparkSession.builder \
    .appName("PySparkDataFrameOps") \
    .config("spark.sql.adaptive.enabled", "true") \
    .getOrCreate()

Create DataFrame

# From a list of tuples
data = [("Alice", 34, 72000.0), ("Bob", 28, 58000.0), ("Charlie", 45, 95000.0)]
schema = ["name", "age", "salary"]
df = spark.createDataFrame(data, schema)
df.show()
# +-------+---+------+
# |   name|age|salary|
# +-------+---+------+
# |  Alice| 34|72000.0|
# |    Bob| 28|58000.0|
# |Charlie| 45|95000.0|
# +-------+---+------+

Transformations (lazy)

# Select, filter, withColumn
df_filtered = df.filter(col("age") >= 30) \
    .select("name", "salary") \
    .withColumn("bonus", col("salary") * 0.1)

# GroupBy + aggregation
df_grouped = df.groupBy("age").agg(
    count("*").alias("count"),
    avg("salary").alias("avg_salary")
)

# Sorting
df_sorted = df.orderBy(col("salary").desc())

# Handling nulls
df_cleaned = df.na.drop()                           # drop rows with any null
df_filled  = df.na.fill({"salary": 50000.0})       # fill specific column

# Add a literal column
df_with_flag = df.withColumn("is_senior", when(col("age") >= 40, lit(True)).otherwise(lit(False)))

Actions (eager)

df.count()          # number of rows
df.collect()        # all rows to driver (careful: memory!)
df.show(5)          # preview first 5 rows
df.take(3)          # first 3 rows as list
df.describe().show() # summary statistics

Reading from various sources

# CSV
df_csv = spark.read.option("header", "true").option("inferSchema", "true").csv("s3://bucket/data/*.csv")

# JSON (nested)
df_json = spark.read.json("s3://bucket/events/*.json")

# Parquet (columnar, schema-preserving)
df_parquet = spark.read.parquet("/data/events/")

# Avro (requires spark-avro)
df_avro = spark.read.format("avro").load("/data/events.avro")

# Kafka streaming
df_kafka = spark.readStream \
    .format("kafka") \
    .option("kafka.bootstrap.servers", "broker:9092") \
    .option("subscribe", "events-topic") \
    .load()

Writing data

df.write.mode("overwrite").parquet("/output/events/")
df.write.mode("append").format("delta").save("/output/delta/events/")
df.write.mode("error").option("header", "true").csv("/output/csv/events/")

UDFs (User Defined Functions)

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

def age_category(age):
    return "young" if age < 35 else "senior" if age >= 60 else "mid"

age_category_udf = udf(age_category, StringType())

df.withColumn("category", age_category_udf(col("age"))).show()

Note: Avoid Python UDFs when possible — they serialize each row row-by-row. Prefer built-in Spark SQL functions (when, regexp_extract, etc.) or Pandas UDFs (vectorized, Arrow-native) for performance.

---

Performance Tuning

Partitioning

Partitions define the unit of parallelism. A partition is processed by one task on one executor core.

# Repartition (shuffles all data — expensive)
df_repartitioned = df.repartition(200, col("country"))

# Coalesce (reduces partitions without full shuffle — use for writing)
df_coalesced = df.coalesce(8)      # from e.g. 200 → 8

# Set shuffle partitions globally
spark.conf.set("spark.sql.shuffle.partitions", 200)

Guidelines: - Target ~128–256 MB per partition after shuffle. - Too few partitions → executors idle (not enough parallelism). - Too many partitions → excessive scheduling/serialization overhead.

Caching & Persistence

Materialise a DataFrame (or RDD) in memory/disk to avoid recomputation.

# Cache (memory only, default MEMORY_ONLY)
df_cached = df.filter(col("age") > 30).cache()
df_cached.count()        # triggers caching

# Persist with storage level
from pyspark import StorageLevel
df.persist(StorageLevel.MEMORY_AND_DISK)   # spill to disk if memory full
df.persist(StorageLevel.MEMORY_ONLY_SER)   # serialised objects (smaller, CPU overhead)

# Unpersist when done
df.unpersist()

Storage levels:

Level	Space Used	CPU Time	In Memory	On Disk
MEMORY_ONLY	High	Low	Yes	No
MEMORY_ONLY_SER	Low	High	Yes (serialized)	No
MEMORY_AND_DISK	High	Low	Yes	Yes
MEMORY_AND_DISK_SER	Low	High	Yes (serialized)	Yes
DISK_ONLY	Low	High	No	Yes

Broadcast Joins

When one side of a join is small (< ~10 MB default), broadcast it to all executors to avoid a full shuffle.

from pyspark.sql.functions import broadcast

# Large table: billions of rows
orders = spark.read.parquet("/data/orders/")

# Small table: a few thousand rows
countries = spark.read.parquet("/data/countries/")

# Explicit broadcast hint (avoids shuffle)
result = orders.join(broadcast(countries), "country_code")

Configurable via:

spark.conf.set("spark.sql.autoBroadcastJoinThreshold", "10485760")  # 10 MB

If the small table is ≤ this threshold, Spark automatically broadcasts it.

Adaptive Query Execution (AQE) — Spark 3.x

AQE re-optimizes the query plan at runtime. Enable it:

spark.conf.set("spark.sql.adaptive.enabled", "true")
spark.conf.set("spark.sql.adaptive.coalescePartitions.enabled", "true")  # default

AQE features: - Dynamic coalescing of shuffle partitions - Dynamic join switching (sort-merge → broadcast if one side is small) - Dynamic skew join handling (splits skewed partitions)

Other tuning tips

Parameter	Recommendation	Effect
spark.executor.instances	Based on cluster size	Number of executors
spark.executor.memory	4–8 GB per core	Heap per executor
spark.executor.cores	2–5 (YARN), 1–5 (K8s)	Parallel tasks per executor
spark.sql.files.maxPartitionBytes	128 MB (default)	Max bytes per partition when reading files
spark.sql.broadcastTimeout	600 (seconds)	Timeout for broadcast build
spark.serializer	org.apache.spark.serializer.KryoSerializer	Faster serialization
spark.shuffle.compress	true	Compress map output files

---

Spark on Kubernetes

Spark 2.3+ natively supports Kubernetes as a cluster manager. Pods are dynamically created per application.

Architecture

• Driver runs as a Kubernetes Pod (not restartable by default → use spark.kubernetes.driver.pod.name)
• Executors run as short-lived Pods, scaled by the shuffle/partition needs
• Cluster manager is the Kubernetes API server — no standalone Master/YARN ResourceManager

Submission Example

$SPARK_HOME/bin/spark-submit \
    --master k8s://https://<k8s-api-server>:6443 \
    --deploy-mode cluster \
    --name spark-pi \
    --conf spark.executor.instances=10 \
    --conf spark.kubernetes.container.image=bitnami/spark:3.5 \
    --conf spark.kubernetes.driver.pod.name=spark-pi-driver \
    --conf spark.kubernetes.authenticate.driver.serviceAccountName=spark \
    --conf spark.kubernetes.namespace=spark-jobs \
    --conf spark.kubernetes.executor.volumes.hostPath.data.mount.path=/mnt/data \
    --conf spark.kubernetes.executor.volumes.hostPath.data.options.path=/data \
    local:///opt/spark/examples/src/main/python/pi.py 100

Best practices on Kubernetes

1. Use dynamic resource allocation: spark.dynamicAllocation.enabled=true (executors scale with workload).
2. Set resource requests/limits — Spark can request specific CPU/memory from Kubernetes:

   --conf spark.kubernetes.executor.request.cores=1.0
   --conf spark.kubernetes.executor.limit.cores=2.0
   --conf spark.executor.memory=4g
   

3. Use node selectors / taints to pin Spark pods to specific GPU or high-memory nodes.
4. Enable local storage with PVs or hostPath for shuffle spill.
5. Node Autoscaling — on cloud (EKS, GKE, AKS), cluster-autoscaler works well with Spark's bursty nature.
6. Monitoring — Spark metrics via Prometheus + Grafana (spark.ui.prometheus.enabled=true).

Spark Operator (Kubernetes-native)

The Spark Operator (by Google) lets you submit Spark applications declaratively via CRDs:

apiVersion: sparkoperator.k8s.io/v1beta2
kind: SparkApplication
metadata:
  name: spark-etl-job
spec:
  type: Scala
  mode: cluster
  image: myregistry/spark:3.5
  mainClass: com.example.ETLJob
  mainApplicationFile: local:///opt/app/etl.jar
  sparkVersion: "3.5.0"
  executor:
    instances: 10
    cores: 2
    memory: "4g"
  driver:
    cores: 1
    memory: "2g"
  restartPolicy:
    type: OnFailure

---

Complete PySpark ETL Example

The following example reads raw JSON events, validates and transforms them, then writes the result as partitioned Parquet. It demonstrates a realistic ETL pipeline with error handling and performance best practices.

#!/usr/bin/env python3
"""
Complete PySpark ETL Pipeline

Source:  JSON event files (S3 / HDFS / local)
Target:  Parquet partitioned by date

Pipeline stages:
  1. Read raw JSON with schema enforcement
  2. Validate fields (null checks, type casting)
  3. Enrich with derived columns (event_date, category)
  4. Filter out invalid or test records
  5. Write partitioned Parquet with mode=overwrite
"""

import sys
from pyspark.sql import SparkSession, DataFrame
from pyspark.sql.functions import (
    col, to_date, when, coalesce, lit, input_file_name, current_timestamp
)
from pyspark.sql.types import (
    StructType, StructField, StringType, LongType, DoubleType, TimestampType, BooleanType
)

# ---------------------------------------------------------------------------
# Schema — enforce at read time to avoid runtime inference errors
# ---------------------------------------------------------------------------
EVENT_SCHEMA = StructType([
    StructField("event_id",   StringType(),   nullable=False),
    StructField("user_id",    StringType(),   nullable=True),
    StructField("event_type", StringType(),   nullable=True),
    StructField("amount",     DoubleType(),   nullable=True),
    StructField("timestamp",  TimestampType(), nullable=True),
    StructField("country",    StringType(),   nullable=True),
    StructField("is_test",    BooleanType(),  nullable=True),
])


def create_spark_session(app_name: str = "PySparkETL") -> SparkSession:
    """Create a SparkSession with performance-optimised settings."""
    return (
        SparkSession.builder
        .appName(app_name)
        .config("spark.sql.adaptive.enabled", "true")
        .config("spark.sql.adaptive.coalescePartitions.enabled", "true")
        .config("spark.sql.adaptive.advisoryPartitionSizeInBytes", "128MB")
        .config("spark.sql.files.maxPartitionBytes", "128MB")
        .config("spark.sql.shuffle.partitions", "200")
        .config("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
        .getOrCreate()
    )


def read_raw_events(spark: SparkSession, input_path: str) -> DataFrame:
    """Read raw JSON events with schema enforcement."""
    print(f"[INFO] Reading events from: {input_path}")
    return (
        spark.read
        .schema(EVENT_SCHEMA)
        .option("badRecordsPath", f"{input_path}/_bad_records")
        .json(input_path)
    )


def validate_events(df: DataFrame) -> DataFrame:
    """
    Validate and clean events.
    - Drop rows missing required fields (event_id)
    - Coalesce null amounts to 0.0
    - Set default country 'UNKNOWN' where null
    - Exclude test records explicitly flagged as is_test = true
    """
    return (
        df
        .filter(col("event_id").isNotNull())
        .withColumn("amount", coalesce(col("amount"), lit(0.0)))
        .withColumn("country", coalesce(col("country"), lit("UNKNOWN")))
        .filter(col("is_test").isNull() | (col("is_test") == lit(False)))
        .drop("is_test")
    )


def enrich_events(df: DataFrame) -> DataFrame:
    """
    Enrich events with derived columns.
    - event_date: date extracted from timestamp
    - event_hour: hour of the event
    - amount_category: buckets for analysis
    - ingest_timestamp: when the pipeline ran
    """
    return (
        df
        .withColumn("event_date",        to_date(col("timestamp")))
        .withColumn("event_hour",        date_format(col("timestamp"), "yyyy-MM-dd HH:00:00"))
        .withColumn("amount_category",
                    when(col("amount") == 0,       "free")
                    .when(col("amount") < 20,      "low")
                    .when(col("amount") < 100,     "medium")
                    .otherwise("high"))
        .withColumn("ingest_timestamp",  current_timestamp())
        .withColumn("source_file",       input_file_name())
    )


def write_parquet(df: DataFrame, output_path: str, partition_col: str = "event_date") -> None:
    """
    Write the DataFrame as partitioned Parquet.
    Uses coalesce — if number of output files is too high, repartition before write.
    """
    print(f"[INFO] Writing partitioned Parquet to: {output_path}")

    # If the DataFrame has many small partitions, coalesce to ~128 MB files
    # Approximate: row_count * avg_row_bytes / 128 MB
    row_estimate = df.count()
    avg_row_bytes = 500  # rough estimate in bytes
    target_files = max(1, int((row_estimate * avg_row_bytes) / (128 * 1024 * 1024)))
    if target_files < df.rdd.getNumPartitions():
        df = df.coalesce(target_files)

    (
        df.write
        .mode("overwrite")
        .partitionBy(partition_col)
        .option("compression", "snappy")
        .parquet(output_path)
    )
    print(f"[INFO] Successfully wrote {row_estimate} rows to {output_path}")


def run_quality_checks(df: DataFrame) -> dict:
    """
    Run basic data quality checks.
    Returns a dict of metrics for logging / alerting.
    """
    total     = df.count()
    null_amt  = df.filter(col("amount").isNull()).count()
    zero_amt  = df.filter(col("amount") == 0).count()
    countries = df.select("country").distinct().count()

    metrics = {
        "total_rows":       total,
        "null_amounts":     null_amt,
        "zero_amounts":     zero_amt,
        "distinct_countries": countries,
    }

    print(f"[QC] {metrics}")
    if null_amt > 0:
        print(f"[WARN] Found {null_amt} rows with null amounts after validation")
    return metrics


def main():
    INPUT_PATH  = sys.argv[1] if len(sys.argv) > 1 else "/data/events/raw"
    OUTPUT_PATH = sys.argv[2] if len(sys.argv) > 2 else "/data/events/clean"

    spark = create_spark_session()

    try:
        # 1. Read
        raw_df = read_raw_events(spark, INPUT_PATH)

        # 2. Validate
        clean_df = validate_events(raw_df)

        # 3. Enrich
        enriched_df = enrich_events(clean_df)

        # 4. Quality checks (cache before action to avoid re-read)
        enriched_df.cache()
        run_quality_checks(enriched_df)

        # 5. Write
        write_parquet(enriched_df, OUTPUT_PATH, partition_col="event_date")

    except Exception as e:
        print(f"[ERROR] ETL pipeline failed: {e}", file=sys.stderr)
        raise
    finally:
        if 'enriched_df' in dir():
            enriched_df.unpersist()
        spark.stop()


if __name__ == "__main__":
    main()

Running the ETL

# Local mode
python3 etl_pipeline.py /data/events/raw /data/events/clean

# Cluster mode via spark-submit
spark-submit \
    --master yarn \
    --deploy-mode cluster \
    --num-executors 20 \
    --executor-memory 8g \
    --executor-cores 4 \
    --conf spark.sql.adaptive.enabled=true \
    etl_pipeline.py hdfs:///events/raw/ hdfs:///events/clean/

---

References

• Official Apache Spark Documentation
• Spark on Kubernetes (SIG)
• Spark SQL Guide
• PySpark API Reference
• Performance Tuning Guide
• Spark Operator (GoogleCloudPlatform)
• Delta Lake Documentation