A pipeline at a company like Flipkart needs to process hundreds of gigabytes of yesterday's transaction logs. On a single machine, that workload can take longer than the available processing window allows, or simply exceed the memory available to a tool like pandas. Spark addresses this by splitting the data into partitions and processing those partitions across many machines at the same time, instead of one machine working through everything sequentially. That difference is not a smarter algorithm. It is **distributed computing** - splitting data across many machines and processing the pieces in parallel. **Apache Spark** is the tool most data engineers reach for to do this kind of large-scale batch processing. > **Note:** The exact speedup from adding more machines is not a clean, > predictable ratio. It depends heavily on the workload, how the data is > laid out, how much shuffling between machines the job requires, network > bandwidth, and cluster configuration. Distributed computing buys you real > headroom on large workloads, but "add more machines, get proportionally > faster" is a simplification worth unlearning early. > 📌 **Remember:** Spark is not a long-term system of record. It reads > persistent data from systems like S3, HDFS, or a warehouse, processes it > across a cluster, and can temporarily hold intermediate or cached data in > executor memory or on local disk along the way - but the durable copy of > your data still lives in whatever system you read from and eventually > write results back to. ### What Spark is not * Not a place to store data permanently - any data it holds in memory or on local disk during a job is temporary, not a durable copy * Not a replacement for a database - it does not serve low-latency point lookups * Not always the right tool - a workload that runs comfortably on one machine rarely needs a cluster > 📌 **Engineering Decision:** Reach for Spark once the *workload* - not > just the raw file size - genuinely benefits from distributed execution. > Warning signs: the data does not comfortably fit in memory on one machine, > a job takes hours instead of minutes and needs to run inside a tighter > window, a transformation is expensive enough that parallelism meaningfully > helps, or the data already lives on a distributed platform your pipeline > reads from anyway. A junior engineer's instinct is often "this file is big, > let's use Spark." The senior engineer's question is "does this workload > actually not fit on one machine anymore?" A 100GB Parquet scan can be > trivial for a single-node engine like DuckDB or a cloud warehouse; a 5GB > job with several expensive joins can strain a laptop. Size alone is a weak > signal - most data engineering workloads in a typical company never > genuinely need a cluster, and pandas or plain SQL in the warehouse stays > the better choice. > 💡 **Practice:** Think of a dataset you have worked with recently, or one > from an earlier module in this roadmap. Would it comfortably fit in memory > on a single laptop, and would its transformations run quickly there? If > yes to both, Spark would be the wrong tool for it - keep that instinct in > mind as you read the rest of this module. **Concept check:** A teammate wants to use Spark to process a 2GB CSV file that easily fits in memory on their laptop. What would you tell them, and why? ---
Every Spark job runs across a small set of cooperating roles. Understanding what each one does is what makes the rest of this module - partitions, shuffles, the Spark UI - actually make sense, rather than feeling like disconnected trivia. Driver | | (splits work, sends tasks) v Executor 1 Executor 2 Executor 3 ... (partition) (partition) (partition) * **Driver** - the coordinator process. It runs your Spark application's `main` code, builds an execution plan, and hands out work to executors. There is exactly one driver per Spark application. * **Executors** - the worker processes that actually process data, one partition at a time, in parallel across the cluster. * **Cluster Manager** - the system responsible for allocating machines and resources to a Spark application (Spark's own Standalone manager, YARN, or Kubernetes). This module treats the cluster manager as infrastructure you configure once, not something you interact with directly in day-to-day PySpark code. ### Jobs, stages, and tasks - how Spark breaks down work Spark does not execute your code line by line the way a Python script does. It builds a plan first. * **Job** - triggered by an action (like writing output or calling `.count()`); one job can contain many stages * **Stage** - a group of work that can run without moving data between machines; a new stage begins whenever data has to be shuffled (see below) * **Task** - the smallest unit of work - one stage's computation applied to one partition, run on one executor Job | +-- Stage 1 (read + filter, no shuffle needed) | +-- Task (partition 1) | +-- Task (partition 2) | +-- Task (partition 3) | +-- Stage 2 (after a shuffle, e.g. a groupBy) +-- Task (partition 1) +-- Task (partition 2) > **Note:** A "stage boundary" is not an arbitrary label - it exists > specifically because moving data between machines (a shuffle) is expensive > and requires the previous stage's output to be fully ready first. Reading > the Spark UI later in this module means reading this exact > job -> stage -> task breakdown to find out where time is actually going. **Concept check:** In one sentence, what triggers a new Spark stage? ---
**PySpark** is Spark's Python API, and DataFrames are the primary way data engineers interact with it - a table-like structure similar in spirit to a pandas DataFrame, except its rows are split across partitions and processed by multiple executors at once rather than living in the memory of a single process. ### Creating a SparkSession Every PySpark application starts by creating a **SparkSession** - the entry point for reading data, running SQL, and building DataFrames. ```python from pyspark.sql import SparkSession spark = ( SparkSession.builder .appName("flipkart_daily_orders") .getOrCreate() ) ``` > **Note:** `getOrCreate()` reuses an existing SparkSession if one is > already running in this process, rather than creating a second one - this > matters in interactive environments like notebooks, where re-running a > cell should not spin up a duplicate session. ### Reading data ```python # Reading with an explicit schema avoids an expensive full-file scan # that Spark would otherwise perform just to infer column types from pyspark.sql.types import StructType, StructField, StringType, DoubleType, TimestampType orders_schema = StructType([ StructField("order_id", StringType(), nullable=False), StructField("customer_id", StringType(), nullable=False), StructField("amount_inr", DoubleType(), nullable=True), StructField("order_ts", TimestampType(), nullable=True), ]) orders_df = ( spark.read .schema(orders_schema) .parquet("s3a://flipkart-lake/raw/orders/") ) ``` > 🔴 **Common Mistake:** Relying on schema inference for a recurring > production pipeline. Schema inference behavior and cost vary by source and > format, but for CSV and JSON in particular, `inferSchema=True` typically > forces Spark to scan some or all of the file just to guess column types > before real work starts. For any pipeline where you already know the > shape of the data - which, for a recurring job, you almost always do - > define the schema explicitly, as shown above. ### Transformations ```python from pyspark.sql import functions as F # select - keep only the columns you need trimmed = orders_df.select("order_id", "customer_id", "amount_inr", "order_ts") # filter - keep only rows matching a condition high_value = trimmed.filter(F.col("amount_inr") > 500) # withColumn - add or replace a column with_flag = high_value.withColumn( "is_premium_order", F.col("amount_inr") > 5000 ) # drop / rename cleaned = with_flag.drop("order_ts").withColumnRenamed("amount_inr", "amount") ``` ### Aggregations ```python # groupBy + agg - total and average order value per customer customer_summary = ( orders_df .groupBy("customer_id") .agg( F.sum("amount_inr").alias("total_spent"), F.avg("amount_inr").alias("avg_order_value"), F.count("order_id").alias("order_count"), ) ) ``` ### Joins ```python # Joining the orders fact data to a customer dimension - the same # star-schema concept from the Data Modeling module, now at scale enriched = orders_df.join( customers_df, on="customer_id", how="left", ) ``` > 📌 **Remember:** Joins are usually the single most expensive operation in > a Spark job, because matching rows across two DataFrames typically > requires moving data between executors so that rows with the same join > key land on the same machine. The Understanding Shuffles section below > explains exactly why, and the Optimisation section shows the main > technique for avoiding that cost when one side of the join is small. ### Writing data ```python ( enriched .write .mode("overwrite") .partitionBy("order_date") .parquet("s3a://flipkart-lake/curated/orders_enriched/") ) ``` > 💡 **Practice:** Using the `orders_df` schema shown above, write a PySpark > transformation that filters for orders placed in the last 30 days, adds a > column classifying each order as `"small"` (under Rs 500), `"medium"` > (Rs 500-5000), or `"large"` (over Rs 5000) using `F.when()`, and groups by > that classification to count orders in each bucket. **Concept check:** Why is `.join()` typically more expensive than `.filter()` or `.withColumn()` in a Spark job? ---
This is where Spark stops being "pandas but distributed" and starts requiring a different mental model. A handful of ideas explain almost every performance problem you will encounter: partitions, shuffling, lazy evaluation, the query plan underneath your code, and how to actually see what your job is doing. ### Partitions - the unit of parallelism A **partition** is a chunk of a DataFrame's rows, and it is the smallest unit of work Spark assigns to an executor. If a DataFrame has 200 partitions and your cluster has 20 executor cores, Spark can process 20 partitions at a time, cycling through the remaining 180 as cores free up. DataFrame (600 million rows) | +-- Partition 1 (3M rows) --> Executor core +-- Partition 2 (3M rows) --> Executor core +-- Partition 3 (3M rows) --> Executor core ... +-- Partition 200 (3M rows) --> Executor core Too few partitions means some executor cores sit idle with nothing to do, wasting cluster capacity you are paying for. Too many tiny partitions means Spark spends more time scheduling and coordinating tasks than actually processing data. Neither extreme is efficient - the right partition count generally scales with how much data you have and how many cores are available to process it. ### Shuffling - the most expensive operation A **shuffle** is Spark physically moving data between executors across the network, so that rows sharing the same key end up on the same machine. It happens whenever an operation needs data grouped or matched in a way the current partitioning does not already provide - most commonly `groupBy`, `join`, and `orderBy`. Before shuffle (data partitioned arbitrarily): Executor A: [cust_1, cust_5, cust_1, cust_9] Executor B: [cust_1, cust_2, cust_5, cust_9] After shuffle (data repartitioned by customer_id): Executor A: [cust_1, cust_1, cust_1] Executor B: [cust_2] Executor C: [cust_5, cust_5] Executor D: [cust_9, cust_9] Shuffles are expensive because they involve writing intermediate data to disk, transferring it over the network, and reading it back in on the receiving executor - all before the actual `groupBy` or `join` computation can even begin. A job that looks simple in code can still be slow if it triggers a large shuffle. > 🔴 **Common Mistake:** Assuming that because a `groupBy` or `join` is > "just one line of code," it is cheap to run. The line of code is short; > the shuffle it triggers underneath can move hundreds of gigabytes across a > cluster network. Always ask what shuffle a transformation will cause, not > just how many characters it took to write. ### Lazy evaluation Spark does not execute a transformation the moment you write it. `.filter()`, `.select()`, `.groupBy()`, and `.join()` all just build up a plan - nothing actually runs until an **action** is called, such as `.write()`, `.count()`, or `.show()`. ```python # Nothing has executed yet after these three lines - Spark has only # recorded the plan filtered = orders_df.filter(F.col("amount_inr") > 500) joined = filtered.join(customers_df, on="customer_id") aggregated = joined.groupBy("city").agg(F.sum("amount_inr")) # THIS line triggers everything above to actually run aggregated.write.parquet("s3a://flipkart-lake/curated/city_revenue/") ``` > **Note:** Lazy evaluation exists so Spark can look at the *entire* chain > of transformations before running any of them, and optimise the whole > plan at once - for example, pushing a filter earlier so less data is > shuffled later. This is also why a typo in a transformation sometimes only > surfaces as an error when you finally call `.write()` or `.show()`, not > when you wrote the line that actually contains the mistake. The next > section shows exactly what Spark does with that recorded plan before any > data actually moves. ### Catalyst, physical plans, and why `explain()` matters Between "you wrote a chain of transformations" and "tasks are actually running on executors," Spark passes your plan through an optimiser called **Catalyst**, and its output through an execution engine called **Tungsten**. You do not need to understand their internals to use Spark well, but knowing this step exists - and how to look inside it - is what separates guessing at optimisations from actually diagnosing a slow job. Your DataFrame code | v Logical plan (what computation should happen) | v Catalyst optimizer (rewrites the plan - e.g. pushes filters earlier) | v Physical plan (how Spark will actually execute it) | v Stages and tasks (what you see in the Spark UI) `.explain()` shows you that physical plan directly, in plain text: ```python enriched = orders_df.join(customers_df, on="customer_id", how="left") enriched.explain(mode="formatted") ``` Reading the output tells you things the Spark UI alone does not make obvious - most usefully, which join strategy Spark actually chose: * `BroadcastHashJoin` - Spark decided one side was small enough to broadcast, no shuffle needed for that join * `SortMergeJoin` - both sides are being shuffled and sorted to match keys, the more expensive default for large-to-large joins * `Exchange` - this line in the plan is a shuffle boundary > 💡 **Tip:** When a join feels slower than expected, running `.explain()` > before touching the Spark UI takes seconds and often tells you immediately > whether Spark chose a `SortMergeJoin` when you expected a broadcast - which > points straight at the fix from the Optimisation section below, before you > spend time digging through stage metrics. The practical debugging workflow this unlocks: check the Spark UI for which stage is slow, check that stage's shuffle bytes to confirm data movement is the problem, then run `.explain()` to see what physical strategy Spark picked and why - before deciding on a fix. ### The Spark UI - reading what actually happened The Spark UI shows the real job -> stage -> task breakdown introduced earlier, with timing and data volume for each stage. The two numbers worth learning to read first: **stage duration** (which stage is actually slow) and **shuffle read/write bytes** (how much data a stage moved across the network). A stage with a long duration and large shuffle bytes is almost always a `groupBy` or `join` that needs the optimisation techniques in the next section - or, as covered next, a sign of skewed data rather than a missing optimisation at all. **Concept check:** You write five chained `.filter()` and `.withColumn()` calls followed by `.write()`. At what point does Spark actually start processing data? ---
Distributed processing is only fast when the work is reasonably balanced across partitions. You can apply every optimisation in this module correctly and still watch a job crawl, because of a different problem: **skew** - one partition holding far more data than the others. Task 1: 2 seconds Task 2: 3 seconds Task 3: 2 seconds Task 4: 47 minutes <- everyone else is waiting on this one A stage does not finish until its slowest task finishes. If 199 tasks process a normal slice of data in seconds each, but one task lands with 40% of the total rows because of a skewed key, the entire stage - and everything downstream of it - waits for that single task. Common causes: a join or groupBy key where one value is dramatically more common than the rest - a `customer_id` of `"guest"` covering millions of unpaired orders, or one wildly popular product or city dominating the dataset while every other key has an ordinary volume. > 📌 **Remember:** Adding more executors does not fix skew. If one partition > holds most of the data, one executor still ends up doing most of the > work, no matter how many idle executors are sitting around waiting for it > to finish. **Diagnosing skew:** in the Spark UI, look at task duration and input size *within* a single stage - not just stage-level totals. A stage where most tasks finish quickly but one or two run dramatically longer, or show a much larger input size, is the signature of skew. **Options, depending on the situation:** * If the skewed side of a join is actually small enough overall, a broadcast join sidesteps the shuffle - and the skew - entirely. * Repartitioning on a different, better-distributed key can spread the load more evenly, at the cost of another shuffle. * **Adaptive Query Execution (AQE)**, covered next, can automatically split Spark's largest skewed partitions in supported situations, without you rewriting the query. * For severely skewed joins that AQE doesn't fully resolve, **salting** - artificially splitting a hot key into several sub-keys to spread its rows across more partitions - is the advanced technique to reach for. It adds real complexity, so treat it as a last resort after the simpler options above, not a default. ---
Everything covered so far - partition counts, join strategy, shuffle behavior - is decided by Catalyst before the job runs, based on estimates. **Adaptive Query Execution (AQE)** lets Spark revisit some of those decisions *during* execution, using real statistics gathered from the data it has already processed, rather than only the estimates it started with. In supported situations, AQE can: * adjust the number of shuffle partitions after seeing actual data volume, instead of using a fixed number decided in advance * switch a join to a broadcast join at runtime if a table turns out to be smaller than expected * automatically split some skewed partitions into smaller pieces, reducing the "one task takes 47 minutes" problem from the previous section > **Note:** AQE is a genuinely useful safety net, not a substitute for > understanding partitions, shuffles, joins, and physical plans. It smooths > over some cases where Spark's initial estimates were wrong, but it does > not turn a fundamentally bad transformation into a good one, and it will > not rescue every skew scenario. Know it exists, know roughly what it can > help with, and keep diagnosing with the Spark UI and `.explain()` as your > primary tools - treat AQE as a second line of defense, not the first. ---
A pipeline at a company like Flipkart needs to process hundreds of gigabytes of yesterday's transaction logs. On a singl...
Every Spark job runs across a small set of cooperating roles. Understanding what each one does is what makes the rest of...
PySpark is Spark's Python API, and DataFrames are the primary way data engineers interact with it - a table-like structu...
This is where Spark stops being "pandas but distributed" and starts requiring a different mental model. A handful of ide...
Distributed processing is only fast when the work is reasonably balanced across partitions. You can apply every optimisa...
Everything covered so far - partition counts, join strategy, shuffle behavior - is decided by Catalyst before the job ru...
Once you can read a Spark UI and physical plan, and know that a shuffle or skew is the likely bottleneck, these techniqu...
Running Spark yourself means managing a cluster. Several managed options remove that burden at different levels of contr...
A teammate hands you this job and says "it worked fine in testing on a sample, but it's been running for two hours on th...
Prerequisites: Python and PySpark installed locally (or access to a Spark cluster / AWS Glue), basic familiarity with SQ...
Concept What it does Key syntax SparkSession Entry point for all Spark operations SparkSession.builder.appName(...).getO...
Relying on schema inference for a recurring production pipeline instead of defining an explicit schema. This happens bec...
Aligns directly with DevOps, Site Reliability (SRE), and Platform Engineering job descriptions.