Big Data Fundamentals

Data Engineering & Modern Data Platforms · Lesson 24 · Big Data Processing and Apache Spark

Lesson Overview

The progress landing in Building Batch Data Pipelines with Python is a correct program. It hashes the closed JSONL file, refuses a prohibited field, quarantines ordinary bad lines under a cap, replaces the published file, and writes the checkpoint last. On a quiet Tuesday that program finishes, the checkpoint moves, and completion-v3 reads one copy of each surviving event_id. Enrollment week is the week the same program is still correct and still late. The planning assumption from Introduction to Data Engineering is about fifty million progress events on a peak day, a few hundred bytes each. The identity map of those ids is the memory cost lesson 14 already named. The morning bound is 06:00 UTC, with the SQL window replace still needing time after the landing. A green design that misses the bound is a failed product for the Monday review.

This lesson is the decision that sits in front of Spark, Hadoop, and every cluster request that arrives the morning after that miss. "Big data" is not a product category and not a reason to adopt an engine. Volume, velocity, and variety are a vocabulary for describing a workload. They do not, by themselves, choose a runtime. The engineering question is narrower: given the contract you already have, the freshness bound you already published, and a measurement of one host, does the work finish with the checks still on? If it does, the lesson 14 job stays. If the key set does not fit and a staging table with a primary key does, that table is the next design, and it is still one system of record for the collision check. If the bytes, the backfill, or the parse rate miss the bound after that, the work has to be split into partitions that fail and retry independently. That split is distributed processing. Apache Spark is one engine that can run it. The engine is lesson 26. The file layout under the split is lesson 25. This lesson decides whether the split is earned.

TechLearningPro's learning analytics estate is the running system. The player still writes progress to PostgreSQL first. The warehouse is still a derived store. The landing still keeps event_id, learner_id, lesson_id, event_type, attempt_number, event_time, and the optional checkpoint integer. Email, name, and payment tokens still fail the file. Distribution does not relax any of those rules. It only changes how many processes may apply them, and what "the published path" means when the output is a directory of Parquet files instead of one JSONL file. Data Formats and Serialization already chose Parquet for scans that should not read every column. This lesson says when that choice becomes the publish format because a single JSONL file is the wrong unit of parallelism, not because a diagram asked for a lake.

Every number below that is not a measurement you took is labeled a planning assumption. Fifty million events, a mean line of a few hundred bytes, and a host size are inputs to a decision. They are not a capacity study and not a vendor benchmark. The skill is to replace each assumption with a measurement before anyone provisions executors.

Learning Objectives

By the end of this lesson you will be able to:

1. Define volume, velocity, and variety as workload descriptions, and separate them from a decision to run a distributed engine.
2. Decide among the lesson 14 single-process job, a relational staging table with a primary key, and a partitioned distributed publish, using fit and a freshness bound.
3. Estimate bytes, key-set memory, and elapsed parse time from stated assumptions, and mark which figures must be measured.
4. Explain data parallelism, a partition as the unit of work and of failure, and the serial work that limits speedup.
5. Choose a partition column whose cardinality matches the publish grain, and reject a high-cardinality column that explodes directories.
6. State which parts of the progress contract, the checkpoint order, and the completion definition must survive a move off one machine.
7. Recognize skew, unsplittable compression, and a too-small sample as reasons a "distributed" design misses 06:00 UTC anyway.

Prerequisites

• Introduction to Data Engineering: the progress-event contract, the six-part lens (correctness, freshness, scale, reliability, security, cost), and the planning assumptions for learners and peak-day events. The warehouse is derived. The player database is the system of record for "where the learner is."
• Data Formats and Serialization: JSONL as a landing encoding, Parquet row groups and file-size targets, and why a million tiny files are an incident.
• Relational Database Internals and /learn/sql: a primary key is a real collision check. You should be able to read a CREATE TABLE and a COUNT.
• Data Integration Patterns and ETL and ELT Architecture: partition replace, a completeness marker after durability, and the placement of completion-v3 in SQL over published facts.
• Building Batch Data Pipelines with Python: hash, refuse, quarantine, atomic replace, checkpoint last. This lesson does not replace that contract. Lessons 4 and 5 and /learn/python are the language for the sizing script.
• A command line and Python 3.11 or newer. The sizing script uses the standard library. The schema probe uses PySpark 3.5 if you have a local session and Java 17. You can read the probe without executing it. No cloud account and no cluster are required.

Detailed Core Concepts

Big is a missed bound or a missed fit

A dataset is operationally large when the design that already preserves correctness cannot meet the freshness bound on the hardware you are willing to run, or cannot hold the state the correctness check requires. That definition is deliberately boring. It does not mention a technology generation. A fifty-million-row extract that streams from disk in half an hour on one host, with a key set that fits in RAM, is a large file and a small job. A five-million-row extract that must be joined, sorted, and published every five minutes, and that misses the five minutes, is a big job even though a slide would call the file small. TechLearningPro's daily mart is the first kind until a measurement says otherwise. The player path is the second kind, and it stays on PostgreSQL. Module 3 does not move the player.

The word "big" became a purchasing adjective. Doug Laney's 2001 framing named three pressures that were already visible to anyone operating warehouses: volume (how much), velocity (how fast it arrives or how fast it must be useful), and variety (how many shapes). Later slogans added veracity and value. Veracity is data quality, which Data Quality Engineering treats as a suite of checks with owners. Value is whether a consumer exists. Those words help a design review ask better questions. They do not compile. A review that stops at "high volume, so Spark" has skipped the bound, the fit, and the contract.

Volume is bytes and cardinality, not a vibe

Volume shows up as two different shortages. The first is storage: the bytes do not fit on the disk you have, or the cost of keeping them on one disk is a decision you have written down. The second is working state: a structure the algorithm needs, such as the set of event_id values seen today, does not fit in memory with the rest of the process. Lesson 14 already separated those. A peak day at a few hundred bytes a line is tens of gigabytes. A modern disk holds that. The Python process that retains a hash per distinct event_id may not, because an interpreter object is much larger than the sixteen or thirty-two bytes of the id. Cardinality, not file size, is what blows the identity map.

A third shortage is throughput. The bytes fit, the key set fits, and the single thread still cannot read, validate, and write before the bound. Throughput is volume divided by the time you actually have, not by the time a brochure quotes. If the job may start only after the producer closes the file at 01:00 UTC, and the SQL job needs the publish by 05:30 UTC, the landing budget is four and a half hours. A design that "finishes overnight" has failed if overnight ended at 02:00.

Backfill volume is a different problem from daily volume. Ninety closed days are not ninety times the operational pain if you can run them sequentially over a weekend and nobody is waiting at 06:00. They are a distributed problem when the business wants the restatement before the next business morning, or when one host would be busy for days and the daily job cannot share it. Write the window down before you write the architecture.

Velocity is the bound, not the arrival rate

Progress events are born all day while learners study. That is a high arrival rate into the player database. The instructor mart does not need those events in the warehouse within seconds. Lesson 1 separated those products on purpose. The mart's velocity requirement is "yesterday is complete before the morning review." A stream processor can meet that bound, and it can also miss it in more interesting ways. Module 4 is where a log earns a place. This module's question is whether the batch landing meets the batch bound. Using the arrival rate as a reason to adopt streaming, while the consumer is a daily grain, spends complexity on a latency nobody requested.

There is a velocity case that does justify leaving one process inside the batch window: the input becomes available late, and the remaining time is shorter than one host's measured runtime. Enrollment week can close the file at 04:00 UTC because the export itself slipped. The bound did not move. The budget shrank. That is a measured squeeze. It is still a batch problem. The repair is a faster correct landing, or a published agreement that the mart is late, not a silent drop of the collision check to make the clock look better.

Variety is a contract problem

TechLearningPro already has more than one shape. Progress is JSONL. A partner seat file is CSV. Enrollment is rows in PostgreSQL. Lesson 11 and lesson 15 assign formats to zones. Variety hurts when one job tries to be a universal parser, or when a new field rides along inside a familiar file. The prohibited-field rule exists because variety showed up as email inside a progress export. A cluster parses the bad file faster. It does not make the file legal. Schema variety belongs to the contract and the zone, which you already have. Distributed execution belongs to fit and the bound. Mixing those two decisions is how a platform team adopts four engines and still publishes two completion rates.

Data parallelism and the partition

When the decision is to split the work, the split this course uses is data parallelism. The same validation function runs on many subsets of the rows. The subsets are partitions. A partition is the unit of work: one task owns it, reads it, and emits a result for it. A partition is also the unit of failure: if that task dies, that subset is redone, and the other subsets stay finished if the engine's plan allows it. Task parallelism, in which different functions run at the same time on the same data, shows up as a pipeline of stages. Both exist in a real job. The landing's first win is data parallelism on the parse. The collision check is a second stage that must see every copy of a given event_id, so those copies have to meet. That meeting is a shuffle. Lesson 30 spends a whole hour on shuffles. Here you only need the cost shape: moving bytes between processes is the expensive part, and an operation that can stay inside one partition is cheaper than an operation that cannot.

A partition is not a business grain. The business grain of the progress fact is one event. The physical partition of a published dataset is a directory or a file group chosen so that reads and retries are tractable. For this landing the publish partition is the UTC event date, derived from event_time after the same offset rules as lesson 14. Analysts filter by day. Restatements replace a day. The cardinality of a date partition is one directory per day you keep, which a catalog can hold. learner_id as a partition column is a different design: on the order of millions of directories, most of them tiny, which is the small-file failure lesson 11 already priced. lesson_id is better than a learner id and still too wide if the catalog stores one partition per lesson and the introductory lesson is a huge fraction of the day. Keep lesson_id as a column. Parquet statistics can skip row groups. The directory layout stays on the date.

Inside a single day the work is split again into tasks. Those task partitions are not Hive-style directories. They are slices of the input, often on the order of a block: 128 MB is the conventional target you will see in both HDFS and Spark's file source, and lesson 25 and lesson 26 separate the two settings. A 20 GB file becomes on the order of a hundred and fifty slices. That number is a parallelism budget. It is also a scheduling budget. Tens of thousands of slices, because each slice is a 200 KB file, spend the morning starting tasks.

What does not parallelize

Amdahl's observation is the restraint on every cluster diagram. If a fraction of the job is serial, the speedup from more machines is bounded by that fraction. In the landing, the serial pieces are real: listing input files, building a plan, the driver's bookkeeping, the final commit that publishes a directory, and the checkpoint write. They are small next to parsing tens of gigabytes, until someone collects every row to the driver "just to dedupe in Python." That collect turns the cluster into an expensive way to fill one process with the entire day, which is the failure lesson 14 already had, plus network. The collision aggregation is parallel, and its result is small only after the aggregation. The driver should see counts and a yes-or-no collision flag, not the events.

A second limit is the input's splitability. Newline-delimited JSON that is uncompressed can be cut on byte ranges, with a reader that finishes the partial line at each boundary. A single gzip member cannot. The codec is a stream. One gzip file of the whole day is one task, no matter how many executors you bought. bzip2 is splittable and expensive in CPU. Many files, each of a sensible size, are the producer's contribution to parallelism. If the producer cannot change, the landing can decompress once into splittable files and then parse. That decompress is a stage with its own bound. Measure it. Do not describe the gzip file as already distributed.

Skew is the third limit. Equal byte ranges are not equal work when one range holds a retry storm of the same event_id, or when the interesting key is lesson_id and one lesson dominates. The stage finishes when the slowest task finishes. A design that looks balanced on a sample of ten thousand lines from a quiet Tuesday is not a design you have tested. Enrollment week's introductory lesson is the production sample that matters.

The contract is not a runtime concern

Correctness travels with the data, not with the engine. These rules are already decided, and a distributed landing keeps them:

• The input identity is the content hash of the closed file, or of the closed set of files if the producer now writes many parts. The name is the address.
• A prohibited field or an unknown field fails the file set. Quarantine of one bad line plus a publish of the rest trains the exporter to keep the column.
• The same event_id with the same canonical contract payload is one event. A different payload fails the run. The loader does not pick a winner.
• event_time is stored in UTC. A timestamp with no offset fails the line. The loader does not assign the business date beyond deriving a partition value from the instant. completion-v3 still owns the window.
• The checkpoint moves after the published dataset is durable and visible. A retry with the same input hash skips only when that publish exists.
• One run of a given product and window at a time, until the scheduler's lock is real. Data Pipeline Orchestration Fundamentals is where that lock starts to exist. Two writers of the same date partition are still a race.

Reliability in the distributed case adds a narrower promise: a lost task redoes its partition from inputs that are still there. It does not invent a second copy of an event. It does not mark the day complete because most tasks succeeded. Security adds a narrower cost: every machine that parses the file holds learner activity in memory and possibly in a spill file. More machines means a wider trust boundary, not a safer one. Cost adds the cluster you pay for while it waits on skew, and the people who have to read a second failure mode at 05:00 UTC.

Three designs, one contract

The decision is a sequence, not a menu of brands.

1. Stay. The lesson 14 job, streaming to a replace, meets the bound, and the identity structure fits on that host. Keep it. A cluster that finishes the same work in ten minutes has spent operability to buy time you did not need.
2. Stage on one database. The key set does not fit the process, and a single database you already operate can take the day's rows with a primary key on event_id and a column for the canonical hash. A conflict on the hash fails the load. The publish step still writes the dataset downstream readers are allowed to see, then the checkpoint. This design uses the engine Relational Database Internals already explained. It is the right middle when the pain is memory of the key set and the clock still fits.
3. Distribute. The measured parse and write miss the remaining budget, or a backfill window misses, or the publish must be many date partitions of columnar files that one process will not write in time. Split the input. Run the same validation per partition. Shuffle only to enforce the event_id rule. Publish by replacing the affected date partitions. Checkpoint last.

Spark is a candidate for the third design because its documented model is partitioned data, lazy plans, and task retry. Hadoop's file system is a candidate for where the bytes live if you already have it, and an object store is a candidate if you do not. Neither sentence is a selection of a vendor product. Lesson 25 and lesson 26 make the mechanisms specific enough to operate. This lesson's output is the written choice, the measurement that justifies it, and the list of contract rules the next lesson is not allowed to drop.

Architecture and Design Diagrams

The decision sits between a closed input and the published dataset completion-v3 may read. Orchestration may start the process. It does not make an unfit process fit.

mermaidCopy
flowchart TD
    closed[Closed progress input]
    measure[Measure bytes, key cardinality, parse rate]
    stay[Lesson 14 process on one host]
    stage[Stage rows with a primary key]
    dist[Partitioned distributed landing]
    publish[Published date partitions]
    mark[Checkpoint after durability]
    sql[completion-v3 reads the publish only]

    closed --> measure
    measure -->|fits memory and bound| stay
    measure -->|key set does not fit, clock does| stage
    measure -->|bound missed or backfill window missed| dist
    stay --> publish
    stage --> publish
    dist --> publish
    publish --> mark --> sql

A distributed landing still has a single visibility rule. Tasks write temporary outputs. Readers of the mart do not see those outputs. A commit makes a date partition appear as a unit. The checkpoint records the input hash only after that commit. A task failure recomputes that task. A failure before commit leaves the previous partition in place.

mermaidCopy
flowchart LR
    parts[Input slices]
    tasks[Validate inside each slice]
    shuffle[Bring identical event_id values together]
    refuse[Refuse the run]
    tmp[Temp task outputs]
    commit[Commit the event_date partition]
    prior[Previous partition remains]

    parts --> tasks --> shuffle
    shuffle -->|conflicting payload| refuse
    shuffle -->|one canonical payload| tmp --> commit
    refuse -.-> prior
    tasks -->|prohibited field| refuse

The placement boundary from lesson 13 does not move because the runtime did. The landing publishes accepted events. It does not compute completion, and it does not drop checkpoint events because a particular aggregation found them inconvenient.

mermaidCopy
flowchart LR
    landing[Landing job, one host or many]
    raw[(Published events by event_date)]
    curated[completion-v3 window replace]
    landing -->|contract fields preserved| raw --> curated
    landing -.->|does not write completion| curated

Technical Explanation

A planning envelope you are required to replace

The figures in this section are a worked envelope so the arithmetic has a shape. They are planning assumptions in the same sense as lesson 1. Replace them with measurements from a real closed file before you spend money.

Assume a peak day of 50,000,000 lines and a mean line length of 360 bytes including the newline. The file is about 18 GB. Assume a measured sustain on the lesson 14 job of 10 MB/s of input while validating. That rate is not a universal Python constant. It is an example of the kind of number a stopwatch produces. At 10 MB/s, 18 GB takes about 1,800 seconds, roughly half an hour, before you add the write of the output and the fsync. A budget from 01:00 UTC to 05:30 UTC holds that run with a wide margin.

Now the key set. Assume nearly every line is a distinct event_id, which matches a clean day of checkpoint events. Holding tens of millions of strings and hashes in a CPython process is commonly several gigabytes to upwards of ten, depending on string length and interpreter overhead. A 16 GB host that is also running the operating system and the file cache may fail that allocation late in the file, after looking healthy on a million-line sample. The clock was fine. The fit was not. The envelope says: do not buy a cluster for the half hour; do something about the key set. A staging table whose primary key is event_id and whose second column is the canonical hash turns the collision check into a constraint the database already knows how to enforce. The copy has to finish inside the same budget. Measure that copy. If it does, design 2 is the landing.

Change one assumption. The measured rate is 1.5 MB/s because each line does more work, or the file is 18 GB of gzip that first expands, or the host is busy. Eighteen gigabytes at 1.5 MB/s is on the order of three hours before the collision structure and the write. The margin against a 04:00 UTC file arrival disappears. Or change the window. A ninety-day restatement is about 1.6 TB at the same mean size. At 10 MB/s, one host needs on the order of two days. If the agreement is "the restatement is queryable tomorrow morning," one host misses it. That miss is a distributed-processing requirement. The daily job and the backfill job are allowed to have different designs. Writing that down prevents a cluster from becoming the daily path by contamination.

What a partition costs

Splitting 18 GB into 128 MB slices yields about 144 input partitions. If those slices can be parsed in parallel, and the machine has the disk and network to feed them, wall time falls. It does not fall by 144. The input device, the output device, and the shuffle for the collision check all have ceilings. A planner who divides 30 minutes by 144 and promises a 12-second landing has ignored every ceiling. A planner who measures one slice, multiplies by a contention factor they state, and then runs a full day once, has an estimate. Spark's default target for a file split, spark.sql.files.maxPartitionBytes, is 128 MB in current 3.5 documentation. That default is a starting point, not a tuning achievement. Lesson 26 shows who reads it. Here it is only the size of a slice.

The collision check cannot stay inside the slice. The two lines that share an event_id may be in different slices, which is likely if the producer appends a retry at the end of the file. Any correct distributed check moves at least the id and the canonical hash, not the whole JSON line, to a common reducer. Moving 50 million short keys is a real shuffle. It is much smaller than moving 18 GB of payloads, and it is not free. Design the record that moves. Do not shuffle the prohibited-field check: that check can happen on the slice, and a single failure fails the run.

File layout for the publish

Readers of the mart should open a columnar dataset partitioned by event_date. Lesson 11's reasons still apply: column pruning, row-group statistics, compression. A single JSONL publish was the right teaching file for one process because os.replace on one path is easy to reason about. Many tasks writing one file is not a good distributed write. Many tasks writing many files under event_date=2026-03-02/ is the layout Spark and similar engines already know how to commit. The visibility rule stays: readers use the committed partition, not the temporary task attempt directories.

How many files inside the day? An 18 GB day of Parquet will be smaller than 18 GB of JSON, often by several times, and the exact ratio depends on cardinality and codec. You measure it. A target on the order of 128 MB to 512 MB per file keeps the next reader from scheduling a task per tiny file and keeps a single file from being a multi-gigabyte straggler. One file for the whole day is legal and often too big for the next job's parallelism. One file per learner is a NameNode or a catalog incident. The publish step chooses the file count on purpose. Lesson 28 implements that choice. This lesson only refuses to leave it to accident.

Failure semantics you can explain without an engine

Use the same crash questions as lesson 14, with the words changed.

• A task dies while reading. That slice is retried. Published data is unchanged. The run has not checkpointed.
• A task dies after writing a temporary file the committer has not promoted. The attempt directory is abandoned. The retry writes a new attempt. Readers of the date partition still see the previous successful day.
• The commit of the date partition succeeds and the driver dies before the checkpoint. The next run repeats the work. The repeat must replace the same date, not append a second copy beside it. If two copies are visible to a glob, the metric doubles, which is the Wednesday retry bug in a new directory layout.
• The checkpoint is written and the input hash matches and the partition is present. The next run skips. If the partition was deleted, the skip is a lie, and the run republishes. Existence is part of the skip rule.
• Two runs of the same date start together. Both may commit. The checkpoint may describe the loser. The operational rule remains one run per product and window.

There is no new exactly-once magic in the word "distributed." There is a commit boundary and a checkpoint order. Engines provide a committer. You still have to know what the committer makes visible, and you still write the checkpoint after that visibility. Lesson 25 talks about why an object-store committer is not an HDFS rename. Lesson 28 keeps the order in code.

Security and cost in the same envelope

An 18 GB progress file is a collection of learner timelines. One host in a private subnet is a trust boundary you can draw. Forty machines, each with a local disk that receives a spill of shuffled keys, is a larger boundary. The keys are event_id and a hash if you designed the shuffle that way, which is better than spilling full payloads, and a hash of a progress event is still operational data. Disks are wiped or encrypted with the settings the engine documents. Logs on those machines record counts, hashes of files, and task ids. They do not record lines.

Cost is the cluster's idle time as well as its busy time, plus the engineering time to operate a second failure mode. A daily job that needs the cluster for forty minutes and a backfill that needs it for six hours can share a pool, or the backfill can be the only distributed job while the daily path stays on the database. Both are legitimate. The illegitimate cost is a permanent cluster sized for enrollment week and empty in August, with nobody allowed to turn it off because the job submission path assumes it is there. Write the calendar next to the architecture.

What this lesson will not pretend to compute

It will not produce a universal threshold such as "above 100 GB use Spark." It will not treat a vendor's autoscaling story as a measurement. It will not move completion-v3 into the landing to "save a pass." It will not drop the collision check because the shuffle showed up on an invoice. It will not call the player database big data because it is important. Importance and distribution are different axes. The player stays a transactional system. The mart stays a derived, replayable dataset with a bound.

Practical Implementation or Design Walkthrough

Problem

You have one closed progress file and a host that already runs the lesson 14 job. Leadership has asked for Spark because the peak-day assumption is fifty million events. You need a written decision: stay, stage, or distribute. The decision uses measurements and budgets, and it names the publish layout if the answer is distribute. A small PySpark session, when you have one, shows that a schema and a partition count are properties of a read, not a reason to skip the decision.

Budgets

Put the budgets in a frozen configuration. The functions do not read the environment themselves. A production runner can build this object from arguments. Thresholds are policy. They belong where a review can see them.

pythonCopy
from __future__ import annotations

import json
from dataclasses import dataclass
from pathlib import Path


@dataclass(frozen=True)
class HostBudget:
    """Planning inputs. Replace every figure with a measurement before you provision."""

    product: str
    ram_gb_for_keyset: float
    landing_budget_seconds: float
    measured_mb_per_s: float
    bytes_per_key_estimate: int
    max_probe_ids: int
    backfill_days: int
    backfill_budget_seconds: float


@dataclass(frozen=True)
class SizingResult:
    file_bytes: int
    lines: int
    distinct_ids_probed: int
    probe_capped: bool
    estimated_keyset_gb: float
    estimated_seconds: float
    decision: str
    reason: str

bytes_per_key_estimate is the planning figure for one id plus hash plus interpreter overhead. Start from a measurement of your own process on a million ids, not from a number you found in a blog. max_probe_ids stops the probe from becoming the out-of-memory incident it is trying to predict.

Probe the closed file without becoming the landing job

The probe streams. It counts bytes and lines. It records distinct event_id values until the cap. It does not publish, does not quarantine, and does not claim to validate the contract. Validation remains the landing job. A probe that re-implements half the contract will drift from it.

pythonCopy
def probe(path: Path, budget: HostBudget) -> SizingResult:
    file_bytes = path.stat().st_size
    lines = 0
    seen: set[str] = set()
    capped = False
    with path.open("r", encoding="utf-8") as handle:
        for line in handle:
            if not line.strip():
                continue
            lines += 1
            if capped:
                continue
            try:
                payload = json.loads(line)
            except json.JSONDecodeError:
                continue
            if isinstance(payload, dict) and isinstance(payload.get("event_id"), str):
                seen.add(payload["event_id"])
                if len(seen) >= budget.max_probe_ids:
                    capped = True
    distinct = len(seen)
    # If the probe stopped early, scale only as a labeled guess.
    if capped and lines > distinct:
        distinct_for_memory = int(distinct * (lines / max(distinct, 1)))
    else:
        distinct_for_memory = distinct
    keyset_gb = distinct_for_memory * budget.bytes_per_key_estimate / (1024**3)
    seconds = (file_bytes / (1024**2)) / budget.measured_mb_per_s
    decision, reason = _decide(budget, keyset_gb, seconds, file_bytes)
    return SizingResult(
        file_bytes=file_bytes,
        lines=lines,
        distinct_ids_probed=distinct,
        probe_capped=capped,
        estimated_keyset_gb=round(keyset_gb, 2),
        estimated_seconds=round(seconds, 1),
        decision=decision,
        reason=reason,
    )


def _decide(
    budget: HostBudget,
    keyset_gb: float,
    seconds: float,
    file_bytes: int,
) -> tuple[str, str]:
    daily_fits_clock = seconds <= budget.landing_budget_seconds
    daily_fits_ram = keyset_gb <= budget.ram_gb_for_keyset
    backfill_seconds = seconds * budget.backfill_days
    backfill_misses = backfill_seconds > budget.backfill_budget_seconds
    if daily_fits_clock and daily_fits_ram and not backfill_misses:
        return "stay", "one host meets the daily bound, the key set, and the backfill window"
    if (not daily_fits_ram) and daily_fits_clock and not backfill_misses:
        return (
            "stage",
            "key set exceeds the process budget; a primary-key stage can own identity if its copy meets the clock",
        )
    if backfill_misses and daily_fits_clock and daily_fits_ram:
        return (
            "distribute_backfill_only",
            "daily job stays; the restatement window does not fit one host",
        )
    return (
        "distribute",
        "daily bound or fit misses after the stated rates; split the input and keep the contract",
    )

The scaled distinct-count when the probe caps is a guess. Say so in the output. A guess that says "maybe ten times memory" is enough to refuse a silent stay. It is not enough to size a cluster. The next measurement is a real count in a system that can hold it, which may be the staging table itself: SELECT COUNT(DISTINCT event_id).

Run the probe on a sample only to check that the script works. Record a second run on a real closed day, or on the largest day you are allowed to read. A decision memo that attaches the sample's elapsed time as if it were the peak day is the anti-pattern this lesson exists to stop.

The schema the distributed path would publish

If the decision is to distribute, the publish schema is the contract, plus the partition column derived from the instant. Nullability on a Spark schema is documentation for the reader. It is not a database constraint. The landing still rejects bad lines in code. The walkthrough uses PySpark 3.5 types so later lessons inherit one schema object instead of three informal lists.

pythonCopy
from pyspark.sql import SparkSession
from pyspark.sql.types import IntegerType, StringType, StructField, StructType

PROGRESS_EVENT_SCHEMA = StructType(
    [
        StructField("event_id", StringType(), nullable=False),
        StructField("learner_id", StringType(), nullable=False),
        StructField("lesson_id", StringType(), nullable=False),
        StructField("event_type", StringType(), nullable=False),
        StructField("attempt_number", IntegerType(), nullable=False),
        StructField("event_time", StringType(), nullable=False),
        StructField("checkpoint", IntegerType(), nullable=True),
        StructField("event_date", StringType(), nullable=False),
    ]
)


def probe_session() -> SparkSession:
    return (
        SparkSession.builder.master("local[2]")
        .appName("tlp-sizing-probe")
        .config("spark.ui.enabled", "false")
        .config("spark.sql.shuffle.partitions", "8")
        .config("spark.sql.session.timeZone", "UTC")
        .config("spark.sql.files.maxPartitionBytes", str(128 * 1024 * 1024))
        .getOrCreate()
    )

event_date is a string yyyy-MM-dd in UTC so the directory name is obvious and does not depend on a session timezone at read time. The source column event_time remains the instant. A local session with two threads is a way to see partitions on a laptop. It is not a test of enrollment week. spark.sql.files.maxPartitionBytes is Spark's documented target for how large a file slice may be. On a small sample every file is one partition because the file is smaller than the target. Print the number and the file size together or the number will lie to you.

pythonCopy
def count_input_partitions(spark: SparkSession, path: str) -> int:
    # Text, not parsed JSON: this probe asks how the file splits, not whether it is valid.
    frame = spark.read.text(path)
    return frame.rdd.getNumPartitions()

A production read that applies PROGRESS_EVENT_SCHEMA to JSON and then writes Parquet partitioned by event_date is the job in lesson 27 and lesson 28. Doing it here would hide the decision under a framework. The schema is in this lesson so the decision memo names columns and partition grain before anyone writes a session.

Memo the decision

A short memo is the deliverable operators can hold. It contains the input path and hash, the probe result, the budget object, the decision string, and what the next lesson is allowed to change. For stay, the next change is none. For stage, the next change is a staging table and a publish query, and the checkpoint still moves last. For distribute or distribute_backfill_only, the next change is a partitioned write of the same columns. The memo states that completion-v3 is unread by this decision. The memo states whether the input is gzip. If it is, the memo says the first distributed slice is a decompress into splittable files, or a producer change, and that buying executors does not split gzip.

What you do not deploy

You do not submit this probe to a scheduler as if it were the landing. It skips validation. You do not log lines it failed to parse. You do not copy the peak-day file to a laptop to make the probe convenient if the file is production learner data. Run it where the landing runs, with the same access scope. The output is counts and a decision string.

Enterprise Real-World Scenarios

Enrollment week, and a cluster request with no stopwatch

TechLearningPro's planning assumption for enrollment week is a peak on the order of fifty million progress events and a Monday review that uses the previous day's completion. The lesson 14 job had been on a single host with 16 GB of RAM. A million-line fixture passed every test. On Wednesday the host was killed by the kernel while building the identity set, after reading most of an 18 GB file. The checkpoint had not moved. The previous publish remained. The mart was stale, which was the correct failure, and the incident channel filled with a proposal to adopt Spark by Friday.

The sizing probe, run on a host that could see the file and capped at a few million ids, reported a capped probe and a scaled key-set estimate above the RAM budget. A one-slice timing of the existing validator produced a rate that put the full file well inside the four-hour budget if the process survived. The decision was stage, not distribute. The squad loaded the closed file into a staging table whose primary key was event_id and whose payload column was the canonical hash, using a database they already operated and already backed up. A hash conflict failed the load. The publish step wrote the accepted events and then the checkpoint, in that order. Thursday's file completed inside the bound. The identity map was no longer a Python object.

The backfill request arrived the same week: restatement of the quarter, about ninety days, before the following morning. Sequential days on one host, at the measured rate, needed longer than the window. The memo's second line was distribute_backfill_only. The daily path stayed on the staging table. The quarter restatement was allowed to use a partitioned engine, with the same refuse rules, writing event_date partitions and checkpointing each date after its commit. Nobody rewrote the daily job onto the cluster "so there is only one path" without a new measurement. Two paths with one contract was the cheaper operability story. One path that missed Thursday to simplify a diagram was not.

Security review asked where the staging rows lived and who could select them. The answer was the same zone as the inbox, not the warehouse role used by instructors. The publish contained the contract columns only. A debug column someone added in the staging table did not get selected into the publish. The cluster, when it ran for the backfill, was scoped to the same zone, and its local disks were in that zone's destruction rule.

Partition by lesson, because that is how the chart is drawn

A second squad accepted a distributed landing and chose the partition column by staring at a dashboard. The dashboard filtered on lesson_id. They wrote one directory per lesson. The introductory lesson, under the planning assumption that checkpoint events dominate and that new learners pile into lesson one, held a large fraction of the day in one partition. Hundreds of other lessons held megabytes. The commit created thousands of directories. The following morning's scan spent more time opening files than reading events. A retry of lesson one dominated the wall clock. The job was distributed and slower than the staging design it replaced.

The repair was a publish partitioned only by event_date, with lesson_id as a normal column inside Parquet files sized to the lesson 11 target. The dashboard's filter became a predicate the reader could push into row groups, which is a property of the format, not a directory per key. File counts inside the day were chosen so each file was a large sequential read. The collision check still shuffled on event_id, where the cardinality is the event, not the lesson. A hot lesson no longer mapped to a single enormous task unless the input split itself was skewed, which the squad then checked by looking at task durations rather than at the dashboard.

The incident note said: a filter column and a partition column answer different questions. The filter column serves a query. The partition column serves replace, retention, and a directory count a catalog can survive. TechLearningPro's retention is by day, because a late correction restates a day and because privacy deletion of a range is discussed as a day-scoped operational problem in later modules. The partition follows the retention and the restatement grain. It does not follow the most interesting chart.

Best Practices

• Write the freshness bound in UTC, including when the input is closed and when the SQL job must start. A decision without a bound is a slogan.
• Measure bytes, line counts, distinct event_id cardinality, and sustained validate rate on a real closed file. Label everything else as a planning assumption.
• Choose stay, stage, or distribute in that order. Revisit the choice when the measurement changes, not when a diagram is redrawn.
• Keep the lesson 14 contract: content identity, file-level refusal for prohibited and unknown fields, collision fails the run, checkpoint after a durable publish, one run at a time.
• Partition published facts by a low-cardinality column you replace and retain as a unit. For this mart, that column is the UTC event date.
• Size files for the next reader. Treat a tiny-file layout as a failed publish even if the row counts match.
• Confirm the input codec can be split before you count executors. gzip of one member is one task.
• Shuffle the smallest record that preserves the rule. The collision check needs event_id and a canonical hash, not the full line.
• Put learner payloads only on machines inside the raw-zone trust boundary. Log counts and file hashes.
• Leave completion-v3 in SQL over the publish. A faster landing that changes the metric is a second product.
• If only the backfill misses the window, distribute the backfill and leave the daily path alone. Record that split so the next review does not "simplify" it away.
• Sample tests prove the script runs. Peak-day tests, or the largest day you may read, prove the decision.

Common Mistakes and Anti-Patterns

Equating the peak-day slogan with a cluster. Fifty million events sounds large in a meeting and may be a half-hour file with a key-set problem. The consequence is a new operational surface that still needs the staging table you skipped, plus a cluster to feed it. The repair is the probe and the three-way decision. Spark remains available when the bound says so.

Collecting the day to one process after paying for many. The job reads in parallel and then gathers every row to the driver to reuse a Python set. The driver dies, the bill remains, and the mart is stale. The repair is an aggregation that returns counts and a collision flag, or a primary key in a database. The driver holds the memo, not the events.

Partitioning by a high-cardinality key. Directories per learner or per lesson make the commit and the next scan slower than the single file you left. The repair is event_date as the only directory partition, with entities as columns. If a single day must be split for file size, split by a salt or by a hash bucket with a fixed small number of buckets, and document the bucket count so readers do not assume one file.

Declaring victory on a sample. Ten thousand lines fit any design. The identity map and the hot lesson appear at full scale. The repair is a probe cap that admits when it extrapolated, and a full-day run before the design is called production.

Dropping the collision check to make the shuffle disappear. Duplicate event_id values with different payloads publish as two facts, or as whichever row a later dropDuplicates kept. Completion moves with no product change. The repair is the same rule as lesson 14: identical canonical payloads collapse, conflicts refuse the run. The shuffle is the cost of that rule. Pay it or stage it in a database. Do not delete it.

Treating an unsplittable file as parallel. One gzip member stays one task. The cluster view shows one executor busy and the others idle. The repair is splittable inputs, many files, or an explicit decompress stage. Idle executors are a clue, not a reason to add more of them.

Moving the metric into the landing because the landing is now "the platform." Checkpoint events disappear, attempt numbers are maxed away, and two squads compute completion differently. The repair is the placement diagram. The landing publishes events. SQL publishes completion.

Using velocity of the player as the warehouse SLA. The player is real-time because a learner is waiting. The mart is daily because an instructor is not. Building a stream for the mart, and operating it through enrollment week, costs more than the landing decision and does not fix a key-set overflow. Module 4 is the place for a log that has a consumer who needs it.

Troubleshooting and Design Review

Symptom: The single-host job is killed late in the file. A small fixture passes. The checkpoint did not move.

Likely cause: the identity structure grew with distinct event_id values until the process exceeded memory. The parse rate may have been acceptable.

Investigation: resident memory over time, line count at death, and the probe's key-set estimate. Confirm the kill is an out-of-memory killer and not a disk-full error during replace. Those two failures have different repairs.

Corrective action: if the clock fits, stage through a primary key and keep publish-then-checkpoint. If the clock does not fit either, distribute the parse and still enforce identity with an aggregation or a constraint. Do not raise the fixture size from ten thousand to twenty thousand and call it a peak test.

Symptom: A distributed run shows one active task for almost the whole wall time.

Likely cause: the input is one unsplittable gzip member, or maxPartitionBytes is larger than the file and there is only one file, or a coalesce to one partition happened before the heavy parse.

Investigation: file count, codec, and the input partition count from a text read. Look at the plan for a shuffle that collapsed to one partition before the expensive map.

Corrective action: change the producer layout or decompress to splittable files. Match the slice size to the file you actually have. Remove the premature coalesce. Adding executors does nothing while the plan has one task.

Symptom: The job finishes and the morning scan is slower than the old single-file path.

Likely cause: the publish created a large number of small files or directories, often because the partition column is lesson_id or learner_id, or because each task wrote a tiny part file per day across many days.

Investigation: count files and directories under the date, and the byte size distribution. Compare task duration histograms if the engine recorded them.

Corrective action: repartition to a deliberate file count inside event_date and stop partitioning the directory tree by entity. Restate one day and compare scan time before you restate the quarter.

Symptom: Distinct event_id in the publish is higher than in the source after a retry.

Likely cause: the commit appended part files to a directory the previous attempt had already committed, or two overlapping runs both committed and a reader globs every part file.

Investigation: checkpoint row, input hash, file listing, and whether the write mode replaces the partition. Look for two commits of the same date in the logs.

Corrective action: stop readers, replace the date from one run, fix the write to replace, and keep the single-runner rule. A downstream DISTINCT hides the double and leaves the directory wrong for the next consumer.

Symptom: The decision memo says distribute, and the first production day is still late, with the collision stage much slower than the parse.

Likely cause: the shuffle moved full payloads, the reducer key was skewed by a retry storm, or the serial commit and the driver did more work than the envelope assumed.

Investigation: bytes shuffled, records in versus records out of the collision stage, and the slowest task's key distribution if you can sample hashes without printing learner ids into a ticket.

Corrective action: shuffle only id and canonical hash. Salt a hot key only if you can still detect a conflict across salts, which means a second small aggregation. If one event_id has millions of copies because of a producer bug, fail the run and fix the producer. Do not size the cluster for a bug.

Symptom: Leadership quotes a three-hour cluster saving against a job that already finished in forty minutes on the staging path.

Likely cause: the comparison used a sample, or compared a warm cluster to a cold host, or ignored operator time.

Investigation: ask for the input hash, the row count, and the bound. Re-run the probe.

Corrective action: keep the design that meets the bound at the lower operational cost. Revisit when the backfill window or the daily rate changes. A faster job that nobody can repair at 05:00 is not an improvement.

A design review for this lesson asks seven questions. What is the bound in UTC? What is the measured rate and the measured cardinality? Which of stay, stage, or distribute did we pick, and which measurement would flip it? What is the directory partition, and what is its cardinality over a year? Can the input be split? What does a reader see if we die before commit? Which contract rules are unchanged? A review that starts by naming an engine has started in the wrong place.

Interview Questions and Answers

Q1. What do volume, velocity, and variety decide?

They describe pressure. Volume is bytes and cardinality. Velocity is how fast data arrives and how fast a named consumer must see a correct result. Variety is the number of shapes and contracts. I use them to make sure a design review asked about all three. I do not use them as a lookup table from adjective to engine. For TechLearningPro's mart, volume is a peak-day file and a key set, velocity is a morning bound rather than the player's request path, and variety is JSONL plus CSV plus a relational enrollment table that we already handle with zones and contracts. The engine decision comes after a measurement against the bound.

Q2. When is one machine still the right daily design at fifty million events?

When a measured run validates and publishes inside the landing budget, and the state required for correctness fits. Fifty million lines of a few hundred bytes are tens of gigabytes. That fits a disk. If the identity structure does not fit memory, I still do not jump to a cluster if a staging table with a primary key can enforce the same collision rule inside the budget. I leave one machine when the clock misses, when the backfill window misses, or when the publish layout itself requires a parallel write we have measured. The slogan size is not the test. The fit and the bound are.

Q3. Why can a dataset fit on disk and still be the wrong shape for the lesson 14 process?

The lesson 14 process streams the file, which addresses disk, and it also retains a hash per event_id, which addresses correctness. Those structures live in the interpreter. Their overhead is far larger than the raw id. A host can have a terabyte free and still kill the process near the end of the file. I measure resident memory against distinct ids. If the clock is fine and the memory is not, I move the set into a database constraint or into an on-disk aggregation. I do not "fix" memory by skipping the collision rule.

Q4. What is data parallelism in this landing?

The same validation runs on many slices of the input at once. Each slice is a partition and a task. A failure retries that slice. Work that can be decided from one line, such as a missing field or a prohibited key, stays in the slice. Work that compares all copies of an event_id cannot stay in the slice, because the copies may not share a slice. That comparison is a shuffle of ids and hashes. Data parallelism speeds the first kind of work. It does not remove the second.

Q5. How do you estimate whether the key set fits?

I count distinct event_id values, or I probe up to a cap and label any scaling as a guess. I multiply by a per-key overhead measured on that runtime, not by the width of the string alone. I compare the product to the RAM I am willing to give the process, leaving room for the file cache and the operating system. If the probe capped, I confirm the cardinality in a system that can hold it before I call the design safe. A million-line sample is a functional test of the counter. It is not an estimate of enrollment week.

Q6. Why is learner_id a poor directory partition for progress events?

A directory partition should be something you replace, drop, and authorize as a unit, at a cardinality a catalog can list. Learners are on the order of millions. Most directories would be tiny, listing them becomes the job, and a restatement of one day would touch every learner who studied that day. I partition by UTC event_date. learner_id stays a column. Privacy deletes and retention policies that are defined per day can then match the layout. A privacy rule that deletes one learner across years is a different, harder operation, and a million learner directories do not make it cheap.

Q7. Which lesson 14 rules survive a move to a cluster?

Content hash as input identity, refusal of the run on prohibited or unknown fields and on payload collisions, quarantine only for ordinary bad lines under a cap, UTC timestamps, no completion logic in the landing, a commit that hides partial output, a checkpoint written after that commit, and a single writer per product and window. What changes is the mechanism: a directory commit instead of os.replace on one file, and a shuffle or a primary key instead of an in-process dictionary. I write the mapping in the memo so a reviewer can see that the promise is the same.

Q8. Where does a serial fraction show up in a landing job?

Scheduling, planning, the driver's collection of task results, the commit that publishes a partition, and the checkpoint write are serial relative to the parse. They are small until someone pulls the full dataset to the driver. The collision shuffle is parallel but has a barrier: the stage does not finish before the slowest task. I look for a gather to the driver first when a cluster is slower than one host. I look for skew and for an unsplittable input when one task dominates. I do not expect a 100-way cluster to produce a 100-way speedup.

Q9. A daily file arrives all day from the player. Why is that not automatically a streaming warehouse?

The player needs a low-latency write because a learner is in the lesson. That write already has a home in PostgreSQL. The warehouse consumer in this course is a previous-day mart with a morning bound. Arrival rate and consumer latency are different requirements. I keep the batch landing until a consumer needs a fresher grain, and that conversation belongs to the streaming module, with its own failure modes for late events. Adopting a stream to rescue an out-of-memory batch job copies the data twice and still needs the collision rule.

Q10. What measurement justifies a distributed landing for TechLearningPro?

A closed input whose validate-and-publish time, at a measured rate, exceeds the remaining budget before 06:00 UTC, after the key-set problem has been given a fair single-system design such as a primary key. Or a restatement window, such as a quarter, that exceeds the time one host can run without violating the agreement. I attach the input hash, the byte size, the rate, the budget, and the codec. I also show that the input splits. A gzip member that cannot split is a producer fix, not a Spark justification. Idle executors are not evidence that I need more of them.

Q11. How can skew hurt a landing that has no join?

Any key that gathers records can skew. The collision check gathers by event_id. A producer bug that retries one id millions of times creates one hot reducer. A parse split that is equal in bytes can still be unequal in CPU if one slice is pathological JSON. A publish partitioned by lesson_id gathers a hot lesson into one directory and often into one heavy task. I read task-duration spread. I keep the collision key, I refuse a broken producer, and I do not pick a directory partition that recreates the hot key as a file layout.

Q12. How does the six-part lens apply to "should we distribute"?

Correctness: the contract, including collisions and prohibited fields, is unchanged. Freshness: the morning bound is the requirement, and a faster wrong publish fails it. Scale: bytes, cardinality, and rate are measured. Reliability: task retry plus commit-then-checkpoint, with one writer. Security: more machines hold learner data, so the trust boundary and spill disks are part of the design. Cost: cluster time, idle time, and operator load have to beat the staging design, not merely beat a laptop. If I cannot fill all six from measurements, I do not have a decision. I have a preference.

Hands-On Exercises

Exercise: Write the stay, stage, or distribute memo

Problem statement. Given a closed progress file, or a synthetic file you generate at a stated scale, produce a decision memo for the TechLearningPro landing. The memo chooses stay, stage, distribute, or distribute-the-backfill-only. It does not skip the contract.

Tasks.

1. State the UTC bound: when the file is closed, when the landing must finish, and what completion-v3 still owns.
2. Implement the probe, or a faithful equivalent, with an explicit cap and a per-key overhead you measured or clearly assumed.
3. Run it on a small fixture to prove the three decision strings can occur, by varying the budget object rather than by weakening validation.
4. Specify the publish schema, including which column is a directory partition and which columns must never be directory partitions. Include one prohibited field in the refusal list.
5. State whether your input codec splits. If you used gzip, explain the consequence for task count.
6. Draw the crash behavior: task death, death before commit, death after commit and before checkpoint. Name what a reader of the date partition sees in each case.
7. Apply the six-part lens in short paragraphs. Label every unmeasured figure.
8. Describe the security boundary of the host or hosts that see raw lines.

Expected deliverables. The probe, the budget object, the memo, the schema, and the crash note.

Evaluation rubric.

You pass when a reviewer can change measured_mb_per_s or the RAM budget and predict the new decision without reading your code a second time, and when the schema's partition column is a date even if the reviewer's chart is about lessons.

Lesson Summary

Volume, velocity, and variety describe a workload. The decision to leave one machine is a comparison of a correct design against a freshness bound and a memory fit. TechLearningPro's progress landing keeps the lesson 14 contract either way. A peak day of tens of gigabytes may be a key-set problem solved by a primary key, a clock problem solved by splitting splittable input, or a backfill problem solved by distributing only the restatement. Directory partitions follow the replace grain, which is the UTC event date. completion-v3 stays downstream.

The next lesson is where bytes sit once a dataset is larger than one disk's operational story: blocks, replication, and the Hadoop ecosystem that made those ideas ordinary, including the ways an object store does not behave like that filesystem. Hadoop Ecosystem and HDFS is that layout. The Spark processes that might compute over it come after, once you know whether you needed them.

Further Reading

• Apache Spark SQL guide, JSON and Parquet files and the Parquet data source — spark.sql.files.maxPartitionBytes and file splits are documented with the reader, not as a slogan. The schema in this lesson is ordinary StructType.
• Apache Spark cluster overview — read it after the decision, when lesson 26 names drivers and tasks. The overview assumes you already have work worth distributing.
• HDFS design — block replication is the storage half of the same split. Lesson 25 walks it.
• Jeffrey Dean and Sanjay Ghemawat, "MapReduce: Simplified Data Processing on Large Clusters" — the paper's contribution is the programming model for data-parallel batch work and the shuffle, which this lesson treated as a cost rather than as a history lecture.
• Martin Kleppmann, Designing Data-Intensive Applications — reliability, scalability, and maintainability as three separate words. The six-part lens is that separation applied to a landing job.
• /learn/python — the probe is standard-library Python. The language course is the syntax reference. Lessons 4 and 5 are the pipeline habits.
• Continue this course: Hadoop Ecosystem and HDFS, then Apache Spark Architecture.