Apache Spark

Repartition in PySparksparkhow repartition, coalesce and column-based partitioning trigger shuffles and when to use each.

1. Spark Basics

What is Apache Spark?
A unified analytics engine designed for large-scale data processing. It provides an interface for programming clusters with data parallelism and fault tolerance.
Core Concepts:
- Resilient Distributed Dataset (RDD): Immutable collections of objects partitioned across the nodes in a cluster.
- Directed Acyclic Graph (DAG): Spark's execution model, representing RDD operations as a series of stages with a lineage of transformations.
- Lazy Evaluation: Spark only executes computations when an action (like collect or save) is triggered.

2. Spark Architecture

Components:
- Driver Program: Manages SparkContext and executes the DAG.
- Cluster Manager: Allocates resources across applications (can be YARN, Mesos, or Kubernetes).
- Executors: Run on worker nodes to execute tasks and store RDD partitions.
- Tasks: Units of work assigned to executors.
Execution Workflow:
1. Submit Job: The driver program initiates a Spark job.
2. Task Scheduling: The DAG scheduler breaks the job into stages, creating tasks for each RDD partition.
3. Execution: Executors perform the tasks and return results to the driver.

3. Spark Components and Libraries

Spark SQL: Enables querying of structured data using SQL or DataFrames.
Spark Streaming: Real-time data processing with micro-batching.
MLlib: Spark's scalable machine learning library for common ML algorithms.
GraphX: API for graph processing and analytics.

4. RDDs, DataFrames, and Datasets

RDDs (Resilient Distributed Datasets):
Basic data structure with operations like map, filter, and reduce. Provides fault tolerance through lineage.
DataFrames:
High-level data structure optimized for querying, backed by Catalyst optimizer. Supports SQL queries, making it a preferred choice for structured data.
Datasets:
Type-safe, object-oriented API available in languages like Scala. Offers benefits of both RDDs and DataFrames but is limited in Python.

5. Key Transformations and Actions

Transformations:
Examples: map, filter, flatMap, union, groupByKey. These are lazy operations that produce new RDDs but don’t execute until an action is called.
Actions:
Examples: collect, count, reduce, saveAsTextFile. Actions trigger the execution of transformations and return results to the driver.

6. Optimizations in Spark

Catalyst Optimizer (for SQL and DataFrames): Uses rule-based and cost-based optimization to produce efficient query plans.

Memory Management: Manages caching, shuffle files, and resource allocation for performance.

How a Spark job executes

Nothing runs until an action such as count() or write is called. At that point the Catalyst optimizer turns the DataFrame operations into an optimised physical plan, and the DAG scheduler cuts it into stages wherever data must be redistributed across the cluster. Each stage runs one task per partition on the executors. Adaptive Query Execution (enabled by default in Spark 3.x) re-plans at stage boundaries using runtime statistics, for example coalescing small shuffle partitions or switching to a broadcast join. On Databricks, the Photon engine can execute many of these operators natively for additional speed.

Choosing an API

API	Use when	Avoid when
DataFrame / Spark SQL	Almost always: optimised by Catalyst, works in Python, SQL, Scala	You need low-level control of partition contents
RDD	Custom partition logic, unstructured records, legacy code	Structured data; you lose Catalyst optimisation
Dataset (Scala/Java)	Compile-time type safety on the JVM	Python (not available)
Structured Streaming	Incremental processing of unbounded data with the DataFrame API	One-off batch jobs
pandas API on Spark	Porting pandas code to larger data	Performance-critical code that needs tuned Spark idioms

Example: inspect a plan

from pyspark.sql import SparkSession, functions as F
spark = SparkSession.builder.appName("demo").getOrCreate()

orders = spark.range(1_000_000).withColumn("cust", F.col("id") % 1000)
custs  = spark.range(1000).withColumnRenamed("id", "cust").withColumn("tier", F.col("cust") % 3)

agg = (orders.join(F.broadcast(custs), "cust")      # avoid shuffling the large side
             .groupBy("tier").count())               # wide: causes a shuffle
agg.explain(mode="formatted")                        # look for Exchange and BroadcastHashJoin
print(agg.collect())

In the plan, each Exchange node is a shuffle and therefore a stage boundary. The broadcast hint ships the small table to every executor so the large table is joined without moving it.

Common pitfalls

Calling collect() on large results, which overloads the driver.
Python UDFs where built-in functions exist; they serialise rows out of the JVM.
Skewed join keys that leave one task running long after the others.
Caching everything; cache only data reused by several actions.
Too few or too many partitions; see the repartition page below.