Hadoop Ecosystem and HDFS

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

Lesson Overview

Big Data Fundamentals ended with a memo. Some days TechLearningPro's progress landing stays on one host. Some days a primary key stages the identity check. Some days, especially a quarter restatement, the input is split and the publish is a set of date partitions. The moment the bytes are expected to survive the loss of a disk, and to be read by more than one process at a time, you need a storage story that is more specific than "a path." Hadoop Distributed File System, HDFS, is the storage system that taught a generation of data platforms what that story is: large files cut into blocks, blocks replicated onto several machines, a single namespace that knows where the blocks are, and a compute layer that used to be MapReduce and is now often something else.

TechLearningPro does not have to build a new Hadoop cluster to learn the model, and this lesson will not pretend the company should. Many lakes now keep the bytes in an object store. Spark can read both. The reason the model is still the lesson is that Spark's file splits, its task attempt directories, and its commit problem are inherited from this design. An operator who thinks "rename the temp directory" is always a metadata operation will publish a torn day on an object store and a clean day on HDFS, with the same PySpark write. An operator who writes one file per progress event will exhaust the process that holds the namespace, whether that process is a NameNode or a catalog listing objects. The ecosystem words around HDFS — YARN, MapReduce, a metastore — are the neighboring processes you will meet on any cluster that already exists, including one a partner still delivers files to.

The running dataset does not change. Closed progress inputs, the contract fields from Introduction to Data Engineering, prohibited fields refused, completion-v3 still computed in SQL over a published dataset as ETL and ELT Architecture placed it. Data Formats and Serialization already chose Parquet for the scan format and warned about small files. This lesson is where those files physically live, how many copies exist after a disk dies, and which process is allowed to remember the directory tree. The checkpoint from Building Batch Data Pipelines with Python still moves after the publish is durable. Durability now means "replicated blocks are acknowledged" or "the object-store committer finished," not "the Python process called os.replace."

Planning numbers stay labeled. An 18 GB day and a 128 MB block are a way to count replicas. They are not a sizing engagement.

Learning Objectives

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

1. Explain an HDFS block, a replica, and the separate roles of the NameNode and the DataNodes.
2. Walk a write and a read, including rack-aware placement and what a lost DataNode changes.
3. Distinguish a Secondary NameNode from a high-availability standby NameNode.
4. Explain why millions of small progress files exhaust namespace memory, and how a date-partitioned Parquet layout avoids that.
5. Place YARN and MapReduce in the ecosystem without treating them as the landing you will ship.
6. Compare an HDFS rename commit with an object-store commit, and keep the checkpoint outside the dataset directory.
7. Estimate block and replica counts for a peak day, and name the operator commands that verify them.

Prerequisites

• Big Data Fundamentals: stay, stage, or distribute, and the rule that directory partitions follow event_date. This lesson is the storage design for the distributed and the simply-large cases. It does not reopen the decision with a new slogan.
• Data Formats and Serialization: Parquet row groups, compression, and file-size targets. HDFS blocks are a different cut from Parquet row groups. You will keep both in your head.
• Building Batch Data Pipelines with Python: publish, then checkpoint. /learn/python is enough language for the arithmetic script. You do not need to have administered a cluster.
• Relational Database Internals: a useful contrast. A database page and an HDFS block both exist to place bytes on disk, and they serve different engines. Do not import SQL isolation vocabulary into HDFS.
• Comfort with a filesystem path and with the idea that a process can die. No Hadoop distribution is required to read the lesson. The commands are the real hdfs dfs client. They do nothing useful without a filesystem to talk to.

Detailed Core Concepts

What HDFS is for

HDFS stores very large files across ordinary machines and keeps serving them when some of those machines are down. The design center is sequential reads and writes of large streams, not low-latency random updates of a single learner's row. The player database remains PostgreSQL for that second job. HDFS, or an object store standing in for it, is a candidate home for the published progress dataset and for a large closed input that no longer fits one operational disk story.

A file in HDFS is a path in a namespace plus an ordered list of blocks. A block is a contiguous piece of the file, 128 MB by default in current Hadoop (dfs.blocksize), not a Parquet row group and not a Spark partition. The last block of a file is as large as it needs to be. A 10 MB file occupies one 10 MB block. It does not consume 128 MB of disk. It does consume namespace memory, which is the small-file problem. Replication, dfs.replication, defaults to 3. Three DataNodes hold copies of that block. The NameNode holds the map. The NameNode does not hold the bytes of the progress file.

That split is the whole architecture. Metadata in one highly available place, bytes spread out, computation moving toward the bytes when the compute layer can ask where a block lives. Object stores keep the metadata/data split and drop the "compute asks for the nearest replica" part, because the store is a service and the compute is often another service. You still need the split in your head to understand both.

NameNode and DataNode

The NameNode is the namespace. It maps paths to inodes: ownership, permissions, modification time, replication factor, and the list of blocks. It maps blocks to DataNode locations. The location map is rebuilt from DataNode block reports when the NameNode starts. The durable metadata is the fsimage plus the edit log. The bytes of part-00007.parquet are not in the fsimage.

DataNodes store block files on local disks, send heartbeats, and send block reports. They also store checksums for chunks of a block so a bad sector can be detected. A client that wants to read does not ask a DataNode to resolve a path. It asks the NameNode for the block locations, then reads the block directly from a DataNode. A client that wants to write asks the NameNode to allocate a block and a pipeline of DataNodes, then streams bytes through that pipeline. The NameNode is in the control path. If it is down and you have no standby, the namespace is down, even if every byte is still on disk. That is why high availability exists, and why "we have three replicas" is not an answer to "the NameNode process died."

A Secondary NameNode, or a checkpoint node, periodically applies the edit log to the fsimage so the log stays bounded. It is a housekeeping role. It does not take over the namespace when the NameNode dies. High availability, added as a first-class layout in Hadoop 2 and still the layout you should expect, runs an active NameNode and a standby NameNode. Edits go to a quorum of JournalNodes. A ZooKeeper failover controller decides which NameNode is active and fences the other. The standby has the namespace in memory and can take over. If someone on an inherited cluster points at a Secondary NameNode and calls it the hot spare, they have named the wrong process. The repair is to look for JournalNodes and a standby, or to admit the cluster is a single NameNode and write that risk down.

Federation, when you see it, means more than one namespace, each with its own NameNode, sharing DataNodes. It exists because one namespace's memory is a ceiling. It is an operator feature, not a landing feature. TechLearningPro's progress tree should be small enough that federation is irrelevant. If it is not, you have too many files, and federation is the expensive way to postpone the cleanup.

Replication, racks, and erasure coding

The default placement policy spreads replicas so one rack is not a single point of failure for a block. The usual statement of the policy is: the first replica is local to the writer when the writer runs on a DataNode, otherwise on a chosen node; the second replica is on a different rack; the third is on the same remote rack as the second, on a different node. Further replicas are placed with an eye on balance. The NameNode can do this only if it knows the rack of each node. A topology script provides that mapping. Without it, every node looks like one rack, and "rack awareness" is a configuration you believe you have.

A lost DataNode stops heartbeating. The NameNode marks its replicas unavailable and schedules re-replication of under-replicated blocks onto other nodes. A lost disk on one node is the same shape, smaller. Three replicas mean you can lose one replica and still read, and you can usually lose one rack and still read, given the placement above. They do not mean you can lose the only NameNode. They do not mean a bug that overwrites a file with wrong bytes is healed: replication copies the bytes you wrote, including a bad publish. Durability is not correctness.

Hadoop 3 adds erasure coding as an alternative to three full copies. A typical Reed-Solomon policy stores a stripe of data blocks plus parity blocks, so the storage multiplier drops from 3 to something like 1.5, and a lost block is reconstructed with CPU and extra reads. That trade is attractive for cold dates of progress you rarely scan and must still keep. It is a poor trade for the file you are rewriting every morning, and it does nothing good for a tree of tiny files, because a stripe wants large sequential blocks. Choose replication for the hot event_date you replace. Consider erasure coding only as a documented policy on cold prefixes, after an operator has enabled it on the cluster. Do not invent a per-job codec and call it erasure coding. Parquet compression, which lesson 11 covered, is a file codec. Erasure coding is a storage layout underneath the file.

The write pipeline and visibility

A client creating a file receives a lease from the NameNode. Bytes flow as packets to the first DataNode in the pipeline, which forwards to the next, which forwards to the last. Acknowledgments return along the pipeline. When a block fills, the client asks for another. When the client closes the file, the NameNode records the final block and releases the lease. Other clients are expected to read a closed file. HDFS can expose data of a file still being written, and that feature is exactly what the landing must not use. The lesson 14 rule, translated: readers of the mart see the previous date partition or the new one, not a block list that is still growing. A commit that renames a temporary directory into the published event_date path is the HDFS-shaped version of os.replace, and on HDFS a rename within one namespace is a metadata operation. It is fast. It is atomic from the reader's point of view in the sense that the destination path appears with its entries. It is not a database transaction across the checkpoint table. The checkpoint is still a later step.

Append exists in HDFS. The landing does not use it. An append of a retried day is the Wednesday double-count, now with replicas.

Small files and NameNode memory

Every file and every block costs metadata on the NameNode heap. The per-object cost is on the order of a hundred bytes plus overhead, and the only number you should trust is a measurement of your NameNode: heap used, files, blocks. The planning shape is enough to scare a bad design. Fifty million progress events written as fifty million files is tens of gigabytes of namespace for one day, before blocks. A peak day as one Parquet dataset, a few hundred files under one date directory, is a few hundred inodes. The first design takes the cluster down for everyone, including jobs that were well behaved. The second design is the lesson 24 publish.

Historical mitigations you may still see: Hadoop archives (HAR), SequenceFiles that pack many small blobs into one file, and combiners in an ingest job. They are answers to a layout that should not have been produced. The repair for TechLearningPro is to refuse the layout. A partner who sends one object per event is a landing problem: compact in the landing, publish Parquet, and do not register fifty million paths in any catalog.

hdfs fsck can show block counts. It is also a load on the NameNode. Use it on a prefix, with a reason, not as a health check every minute.

YARN, MapReduce, and what you are not shipping

HDFS does not schedule your program. YARN does, on a classic Hadoop cluster. The ResourceManager assigns resources. A NodeManager on each machine runs containers. Each application has an ApplicationMaster that asks the ResourceManager for those containers and tracks them. MapReduce was the first application: a map stage, a shuffle, a reduce stage, with the ApplicationMaster running the job. The collision rule can be drawn that way. Map reads a line of event_id and canonical hash, already produced by a trusted parser. Reduce sees every hash for one event_id. If two hashes appear, it emits a conflict. The driver fails the job if the conflict counter is non-zero. That is a correct sketch of data-parallel batch work. It is also how many inherited jobs on a partner cluster still look.

TechLearningPro's landing, if it distributes, will be the PySpark job in later lessons, not a new MapReduce codebase. MapReduce materializes the shuffle to disk between stages and offers a narrower API. Spark, lesson 26, keeps a more general plan and can hold intermediate data in memory when it fits. You learn MapReduce so you can read the old job and so the words "shuffle" and "slot" on that cluster mean something. You do not port completion-v3 into a reducer. Completion stays SQL over the published files. YARN may still be the cluster manager that launches Spark. Kubernetes may be. The standalone Spark manager may be. Lesson 26 compares those. Here, YARN is the resource layer that made HDFS and a compute framework share a fleet, with queues so the landing does not consume every container the cluster has. The Capacity Scheduler is the default scheduler class in current Hadoop. A queue with a capacity is an operability tool: the progress landing gets a budget, and a runaway backfill does not eat the NameNode's neighbors.

Object stores are not HDFS with a different hostname

An object store keeps bytes under a key and replicates them inside the provider's durability model. You do not pick a rack. You do not run a NameNode. Listing a prefix is an API call that can be slow and paginated. A "rename" of a directory of part files is often implemented by the Hadoop compatibility layer as copy and delete, not as a metadata pointer swing. Spark on HDFS can commit a task attempt by renaming _temporary/attempt/... into the output directory. The same code path against s3a://, which is the Apache Hadoop S3A filesystem, can turn that rename into a large copy, fail partway, and leave a date partition with a mixture of old and new part files. Hadoop documents committers for this reason, including a magic committer that writes directly to the final object name via a multipart upload so the object appears when complete. The configuration keys live in the Hadoop AWS documentation for the version you run (fs.s3a.committer.name and the related Spark commit protocol settings). This lesson names the problem and the documented family of fixes. It does not invent a TechLearningPro committer.

The contract translation is short. On HDFS, a rename commit plus a checkpoint after it matches lesson 14. On an object store, use the committer the vendor documents for that filesystem, confirm a failed job does not leave a readable partial partition, and only then write the checkpoint. A _SUCCESS file, written by Hadoop's output committer, means the job's output commit finished. It does not mean your collision check passed, unless your job failed before commit when the check failed. It does not replace the checkpoint row that records the input hash. Put _SUCCESS where the committer puts it, inside the output, and put the business checkpoint outside the dataset root so a reader that globs event_date=*/part-* does not also glob a JSON control file into a schema surprise. Some readers ignore unknown files. Do not rely on that.

Permissions and the raw zone

HDFS permissions look like Unix owner, group, and mode, with optional ACLs. Authentication on a real multi-tenant cluster is Kerberos, and jobs receive delegation tokens so executors can act for the submitter. Encryption zones can encrypt a prefix at rest with keys from Hadoop's key management server. You configure those with the cluster's documented procedures. What this lesson insists on is the zone boundary from lesson 13: the progress prefix is readable by the landing role and by the curated job's role, not by the instructor warehouse role that should see completion-v3 and not the raw event. Replication does not change the ACL. Three copies with a world-readable mode are three copies of a leak. A partner dump that arrives on an inherited HDFS cluster inherits that cluster's authentication. Copying it to a laptop to "just look" is the incident.

Executor spill and shuffle files from lesson 24's security note are not HDFS blocks unless you configured a shuffle service on HDFS, which is unusual. They are local disks. HDFS replication does not encrypt those local disks for you. Say so in the run book so nobody points at dfs.replication as the privacy control.

Architecture and Design Diagrams

A closed input becomes blocks only after a client is told where to put them. The NameNode never sits on the byte path.

mermaidCopy
flowchart LR
    client[Landing client or Spark task]
    nn[NameNode namespace]
    dn1[DataNode rack A]
    dn2[DataNode rack B]
    dn3[DataNode rack B]
    reader[completion-v3 reader]

    client -->|create and allocate block| nn
    nn -->|pipeline locations| client
    client -->|packets| dn1 --> dn2 --> dn3
    dn3 -->|acks back along the pipeline| client
    client -->|close file| nn
    reader -->|block locations| nn
    reader -->|read bytes| dn1

The publish layout keeps namespace objects few, and it keeps the checkpoint off to the side. A reader of the dataset is pointed at progress/. A reader of the checkpoint is the landing's own runner.

mermaidCopy
flowchart TD
    root[progress dataset root]
    day[event_date=2026-03-02]
    parts[Parquet part files, hundreds not millions]
    success[_SUCCESS from the committer]
    ckpt[Checkpoint record outside the dataset root]
    sql[SQL reads part files under the date]

    root --> day --> parts
    day --> success
    ckpt -->|written only after commit| day
    parts --> sql

YARN sits beside HDFS, not inside it. MapReduce is one application. Spark is another. The landing you ship is the Spark application, later. The picture stops the two from being confused.

mermaidCopy
flowchart LR
    hdfs[(HDFS blocks)]
    rm[YARN ResourceManager]
    am[ApplicationMaster]
    nm[NodeManagers and containers]
    mr[Inherited MapReduce job]
    spark[Spark application, later lessons]

    rm --> am --> nm
    nm -->|read and write blocks| hdfs
    am -.-> mr
    am -.-> spark

Technical Explanation

Blocks, splits, and row groups

Three cuts are easy to mash together. An HDFS block is the replication unit, default 128 MB. A Spark input split is the unit of one task. On HDFS, splits are typically aligned to blocks, and Spark's file source also respects spark.sql.files.maxPartitionBytes, default 128 MB, so the numbers often match until someone sets one of them and not the other. A Parquet row group is the compression and statistics unit inside the file, often smaller or similar, chosen by the writer. A task may read one or more row groups. A block may contain several row groups or, if you mis-size the writer, a row group may straddle a block boundary and a scan reads two blocks to assemble it. The practical rule: write Parquet files on the order of one or a few blocks, with row groups that do not dwarf the reader, and let one task read a coherent slice. Lesson 28 sets the writer. This lesson forbids a design that treats "block," "file," and "partition directory" as synonyms.

Checksums are per chunk inside the block, historically 512 bytes. A DataNode verifies them. A client can verify them. A silent bit flip is a storage incident, repaired from another replica if one exists. It is not a business-level validation. A prohibited email field replicates perfectly. The contract check still runs in the landing, before commit.

How much namespace a peak day costs

Use the planning assumption of about 18 GB of JSON input and a smaller Parquet publish, and a 128 MB block. Eighteen gigabytes is about 144 blocks. At replication three, the cluster stores about 432 block replicas, plus pipeline space while the write is open. As namespace, that is one file or a few hundred part files, plus one directory for the date. A NameNode that can hold hundreds of millions of files, which is an operator conversation about heap, does not notice a few hundred. The same day written as one file per event is fifty million inodes and at least that many blocks. That is a cluster-wide incident: slow listings, long checkpoints of the fsimage, and a startup time measured in many minutes or worse. The arithmetic script in the walkthrough prints both layouts so the memo can show the bad one next to the good one.

List performance follows the same arithmetic. Listing a date directory of two hundred files is a small RPC. Listing a day of fifty million files is a denial of service against the NameNode, and the Spark driver that tries to list it will look like a slow job before it looks like a storage incident. If the driver spends the morning in "listing input," count the files before you tune executors.

Re-replication and the morning job

A DataNode that dies during the night causes re-replication work in the morning, which competes with the landing for disk and network. Three replicas make the file readable while that work happens, as long as one replica of each block remains. The landing's own write also uses the network: a client sends one copy, and the pipeline writes the other two, so the cluster network carries more than the logical 18 GB. Budget the pipeline, not just the logical size, when the cluster is busy rebuilding. This is a cost and reliability interaction. It is not a reason to set replication to one so the job looks faster. Replication one means a single disk failure deletes blocks of learner history. If the prefix is an object store, the provider's durability replaces this paragraph, and you read that provider's documented durability instead of inventing an equivalent of dfs.replication.

The inherited MapReduce shape, briefly

You will sometimes find the collision check already written as MapReduce on a partner cluster. Map output is partitioned by hash of event_id so one reducer sees all hashes for that id. The reducer writes either nothing or a conflict record. Counters report the conflict count. job.waitForCompletion(true) returns, and the driver process inspects counters. A non-zero conflict counter fails the application before anything is renamed into the published date. That order is the contract. A reducer that picks "the latest hash" violates the contract even if the Java compiles. When you touch such a job, preserve the fail-the-run behavior and plan a port to the DataFrame job only when lesson 28's tests cover the same fixtures: identical duplicate, conflicting duplicate, prohibited field.

Speculative execution, present in MapReduce and in Spark, runs a second copy of a slow task. It is safe when tasks are deterministic and side effects happen through the committer, which ignores the losing attempt. It is unsafe when a task appends to an external system on its own. The landing's tasks write attempt files only. They do not send mail and do not update the checkpoint. The driver updates the checkpoint once. That discipline is what makes speculation and task retry legal.

Catalogs

A directory tree can be the whole product interface: the SQL engine is pointed at the root and partitions are discovered from directory names. A metastore, historically the Hive metastore, stores a table name, a schema, and a location, and a list of partitions if you add them. Spark can read the path with an explicit schema and never talk to Hive. If a metastore is updated by a separate process that lags the commit, readers see the old partition list and miss the day you just wrote. Prefer a commit that either registers the partition in the same operational step as the checkpoint or uses a reader that discovers directories from the filesystem and does not trust a stale catalog. Later modules cover table formats that combine files and a catalog in one commit. This lesson's rule is enough: the files that exist after commit are the data, the checkpoint says which input produced them, and a catalog that disagrees is a bug to fix, not a second truth.

Practical Implementation or Design Walkthrough

Problem

Design the storage layout for the published progress dataset and a way to estimate its cost on HDFS. Show the commands an operator runs against a real cluster. Show the PySpark read path a later job will use. Provide a short, real MapReduce skeleton for the inherited collision check on event_id and hash, so a partner job is readable. Do not claim the skeleton is the daily landing.

Layout

textCopy
hdfs:///data/tlp/raw/progress/event_date=YYYY-MM-DD/part-*.parquet
hdfs:///data/tlp/raw/progress_checkpoint/<product>/<input-name>.json

The checkpoint prefix is not under progress/. The dataset root contains only date partitions. Permissions: the landing principal owns the prefixes. The curated SQL principal can read progress/ and cannot read the inbox. The instructor role reads neither. Replication is 3 on the hot prefix unless an operator has a written erasure-coding policy for dates older than the retention hot window.

Arithmetic you can rerun

The script uses the standard library. It does not connect to HDFS. It makes the replica count obvious before anyone files a ticket.

pythonCopy
from __future__ import annotations

from dataclasses import dataclass


@dataclass(frozen=True)
class BlockEstimate:
    logical_bytes: int
    block_bytes: int
    replication: int
    file_count: int

    @property
    def blocks(self) -> int:
        # Each file has at least one block. A file smaller than a block is one block.
        full = self.logical_bytes // self.block_bytes
        remainder = 1 if self.logical_bytes % self.block_bytes else 0
        by_size = full + remainder
        return max(by_size, self.file_count)

    @property
    def replicas(self) -> int:
        return self.blocks * self.replication


def compare_layouts(logical_bytes: int) -> tuple[BlockEstimate, BlockEstimate]:
    block = 128 * 1024 * 1024
    # A few hundred Parquet files, one date directory. file_count floors the block count.
    columnar = BlockEstimate(
        logical_bytes=logical_bytes // 3,
        block_bytes=block,
        replication=3,
        file_count=200,
    )
    one_file_per_event = BlockEstimate(
        logical_bytes=logical_bytes,
        block_bytes=block,
        replication=3,
        file_count=50_000_000,
    )
    return columnar, one_file_per_event

The logical_bytes // 3 factor is a placeholder for "Parquet was smaller in a sample," not a law. Replace it with the size of a real event_date directory. The file count of fifty million is the anti-pattern, and blocks becomes at least fifty million because each file has a block. Print both replica totals in the design memo. The columnar number should be hundreds of replicas. The per-event number should be refused.

Commands that are real on a cluster

These do nothing in the lesson environment unless hdfs is configured. They are the operational interface, not pseudocode.

textCopy
hdfs dfs -ls /data/tlp/raw/progress
hdfs dfs -du -h /data/tlp/raw/progress
hdfs dfs -stat "%n %b %o %r" /data/tlp/raw/progress/event_date=2026-03-02
hdfs dfs -df -h /
hdfs fsck /data/tlp/raw/progress/event_date=2026-03-02 -files -blocks

%o is the block size and %r is the replication, from the hdfs dfs -stat format documented with the CLI. fsck on a single date is a diagnostic. fsck on / during the landing window is how you make the NameNode busy at the worst time. -du walks the tree. On a healthy date partition it is cheap. On a fifty-million-file prefix it is another incident. If -ls on the date is slow, stop and count files. Do not add executors.

How a later Spark job names the path

The read uses the nameservice, which fails over with the NameNode, rather than a single hostname that is down during HA. The schema is the contract schema from lesson 24, without adding columns HDFS does not know about. HDFS has no schema. The schema is the file format plus what you declare.

pythonCopy
from pyspark.sql import SparkSession

def read_progress(spark: SparkSession, nameservice: str):
    root = f"hdfs://{nameservice}/data/tlp/raw/progress"
    return (
        spark.read.schema(PROGRESS_EVENT_SCHEMA)
        .option("basePath", root)
        .parquet(root)
    )

PROGRESS_EVENT_SCHEMA is the StructType from lesson 24. basePath tells the reader that event_date is a partition column in the path. On an object store the same call uses s3a://tlp-raw/progress and depends on the S3A committer when it is time to write. Reading is the easy half. Writing is where HDFS and the object store diverge, and lesson 28 keeps the write next to the checkpoint order. A local SparkSession cannot resolve hdfs:// without a configuration that points at a real cluster. Do not mock a NameNode in unit tests. Test the layout arithmetic and the schema locally. Test hdfs dfs -stat in an environment that has HDFS.

An inherited reducer you can read

The input is already event_id and a canonical hash, tab-separated, produced by the trusted parser. This is not a JSON parser and not the landing. It shows the shuffle key and the refusal.

javaCopy
import java.io.IOException;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

public class CollisionReducer extends Reducer<Text, Text, Text, Text> {
  private int conflicts;

  @Override
  protected void reduce(Text eventId, Iterable<Text> hashes, Context context)
      throws IOException, InterruptedException {
    String canonical = null;
    for (Text hash : hashes) {
      String value = hash.toString();
      if (canonical == null) {
        canonical = value;
      } else if (!canonical.equals(value)) {
        conflicts++;
        context.getCounter("quality", "collisions").increment(1);
        context.write(eventId, new Text("collision"));
        return;
      }
    }
  }
}

The driver, after waitForCompletion, reads the quality / collisions counter and fails the process if it is non-zero, before any rename into /data/tlp/raw/progress. Identical hashes emit nothing. That matches the contract: duplicates of the same payload collapse, a second payload refuses the run. A Java class that resolves the conflict by timestamp is a contract bug. When you port this to PySpark, the test fixtures move with it. The reducer does not see email, because the parser upstream already refused the file. If this job is the first code that touches partner JSON, it is the wrong job. Put the contract check where the file is first accepted.

Operational notes

• One writer per date. YARN does not know that two Spark applications targeting the same directory are enemies. The scheduler must.
• Checkpoint after the commit, outside the dataset root.
• Hot dates stay at replication 3. Cold dates move to an erasure-coding policy only through the cluster's documented tool, with a measured read penalty.
• The landing principal's keytab or token is a secret. It does not appear in the memo, the log, or the Parquet footer.
• A disk failure during the run is a DataNode problem if the pipeline can replace the node, which HDFS is built to do. A NameNode failover is a pause, then success, if HA is real. A single NameNode reboot is an outage. Know which cluster you have before enrollment week.

Enterprise Real-World Scenarios

The partner cluster and fifty million inodes

A content partner delivered "HDFS access" to TechLearningPro as a tree of progress-like files, one JSON file per event, on a cluster they administered. The first morning an engineer listed the prefix from the driver of a trial Spark job. The list did not finish before the maintenance window. hdfs dfs -count on a subdirectory showed file counts in the millions for a single day, consistent with the planning assumption of a peak day and a layout of one file per event. The NameNode heap was already the partner's ongoing incident. Adding TechLearningPro's scan made their other jobs slow. The data was durable. The namespace was the outage.

The repair was a compacting landing that ran on the partner cluster with a queue the Capacity Scheduler could cap. It read the small files in batches the NameNode could list, wrote Parquet under event_date with a few hundred part files, and wrote a checkpoint outside that root. Downstream SQL was pointed only at the compacted tree. The small-file tree remained the partner's raw delivery, with a retention clock, and was not registered as a table. A proposal to turn on NameNode federation so the small files could stay was declined. Federation would have added a second namespace for a layout the contract did not want. The partner was asked to change the delivery. Until they did, the compacting job was the isolation layer, and its queue limit was what kept the NameNode available to everyone else.

Security followed the files. The partner prefix was not added to the instructor role. The compacted Parquet contained the contract columns after the prohibited-field check. A sample that included email failed the compact and left the previous date in place. The incident write-up counted inodes, not rows, as the primary damage metric.

Rename commit on an object store during a restatement

The quarter restatement from lesson 24's second decision line wrote date partitions to an object store using the same Spark overwrite the team had tested on a small HDFS namespace. On HDFS, the commit renamed attempt directories. On the object store, the compatibility layer copied each part file into place. A ninety-day restatement multiplied that copy. The job missed the window. One date failed midway through the copy, and readers who listed the prefix saw a mixture of new and old part files. _SUCCESS was absent for that date, but a SQL engine pointed at the prefix with a glob did not check _SUCCESS. The completion number for that day moved halfway, then broke.

The repair had three parts, all documented behavior rather than a new framework. The output committer was set to the S3A committer the Hadoop version documented for that bucket, and a failed job was re-run to show that readers either saw the previous complete partition or the new one. The reader was changed to require the checkpoint row for that date before scanning, which is the lesson 14 rule and does not depend on _SUCCESS. The checkpoint stayed outside the dataset prefix. After that, the restatement time was dominated by the parse and the shuffle, which had been the estimate, not by a hidden second copy of every byte. The HDFS test environment was kept for contract tests. It was no longer treated as proof of the object-store commit. The run book says which committer each prefix uses. A new bucket does not inherit the assumption.

Best Practices

• Store published progress as Parquet under event_date=yyyy-MM-dd, with hundreds of files in a peak day, not one file per event.
• Keep the business checkpoint outside the dataset root. Treat _SUCCESS as a committer signal, not as the checkpoint.
• Use replication three for hot date partitions unless a written erasure-coding policy covers cold data.
• Learn the difference between a Secondary NameNode and an HA standby before you describe the cluster as highly available.
• Configure rack topology if you depend on rack-aware placement. An unconfigured script means one logical rack.
• Point Spark at a nameservice, not at a single NameNode host, when HA is enabled.
• On object stores, use the documented committer. Retest visibility on failure. Do not treat an HDFS rename test as sufficient.
• Cap the landing's YARN queue so a backfill cannot starve the cluster or the NameNode.
• Run fsck and full-tree listings as diagnostics, not as polling.
• Refuse prohibited fields before commit so bad columns are not replicated three times into the raw zone.
• Leave MapReduce jobs you inherit fail-closed on hash collisions. Do not "fix" them to keep the latest record.
• One writer per date, enforced by the scheduler, because the filesystem will not enforce it for you.

Common Mistakes and Anti-Patterns

Calling the Secondary NameNode a hot standby. It merges fsimage and edits. When the active NameNode is gone, a Secondary NameNode does not serve the namespace. The consequence is an outage you described as covered in a design review. The repair is JournalNodes, a standby NameNode, and a failover test. Until that test passes, the run book says the namespace is a single process.

One file per event, or per learner. Namespace memory and list time dominate. Replicas multiply the metadata as well as the bytes. The repair is compaction to date-partitioned Parquet before any catalog registration. HAR files are a last resort for an inherited tree you cannot rewrite yet, not a target layout.

Append as retry. HDFS will append if you ask. A retried day then contains two copies, replicated three times, and SQL counts both. The repair is a replace of the date directory through the committer, with the input hash in the checkpoint.

Assuming rename is cheap everywhere. On HDFS it is a metadata operation. On an object store the same API can be a copy. The consequence is a missed window and a mixed prefix. The repair is the documented committer and a reader that trusts the checkpoint.

Setting replication to one to save money on the hot prefix. A single disk loss deletes blocks. The landing looks cheaper until the first failure, and the failure deletes learner history that the player database may already have aged out of the analytical extract. The repair is replication three, or the object store's documented durability, and a cost conversation that includes restore.

Reading an unclosed file from the SQL job. HDFS can expose bytes before close. The mart then sees a partial day and a later completion jump. The repair is the same visibility rule as lesson 14: only the committed partition, and only after the checkpoint. Discovery jobs do not point at _temporary.

Putting the checkpoint JSON inside the date directory. A schema-on-read scan tries to parse it as Parquet, or a glob picks it up, or a drop of the date deletes the only record of the input hash. The repair is a separate prefix and a reader that never globs it.

Using fsck as a heartbeat. It walks namespace state and competes with the landing. The repair is DataNode heartbeats and disk metrics for liveness, and fsck when you suspect missing blocks.

Troubleshooting and Design Review

Symptom: The landing is slow and the driver log spends a long time listing paths. Executor count does not change the wall clock.

Likely cause: too many files under the input prefix. The NameNode is serving a huge listing. Tasks have not started in earnest.

Investigation: hdfs dfs -count on the prefix, off peak if the count itself is dangerous. NameNode heap and RPC latency. File-size histogram.

Corrective action: compact first. Do not scale executors into a list storm. If the partner cannot change delivery, isolate the compact job in a capped queue and publish only the compacted tree.

Symptom: A date directory contains part files from two different input hashes, and completion moved twice.

Likely cause: append, or a commit that copied new files beside old files, or two applications writing one date.

Investigation: checkpoint contents, _SUCCESS timestamp, part-file names, and YARN application ids for that window. Compare a sample of file footers or row counts with the checkpointed hash's expected counts.

Corrective action: stop readers, rewrite the date from one input, fix the committer or the single-writer rule, and only then move the checkpoint. Do not delete "extra" part files by hand unless you can prove which attempt they came from.

Symptom: Reads fail with missing blocks after a rack outage, even though replication was set to three.

Likely cause: replicas were not actually rack-spread, because topology was unset, so one rack held all three. Or replication was overridden to one on that path. Or the outage lasted long enough and took more replicas than you had, and re-replication had not finished.

Investigation: hdfs dfs -stat for %r, fsck on the date for missing replicas, and the topology script's output for two nodes that should be in different racks.

Corrective action: restore topology, confirm placement on a new file, and re-replicate. For the landing, fail the morning read rather than publish a partial scan as a complete day. A partial scan that silently drops a block is a wrong metric.

Symptom: NameNode failover "exists" on a slide, and a NameNode restart caused a full landing outage.

Likely cause: the process you restarted was the only active NameNode. The other process was a Secondary NameNode, or the standby was not actually caught up, or fencing failed and failover refused to proceed.

Investigation: which daemons are JournalNodes, which NameNode is active, and whether a failover test has ever been done outside a slide.

Corrective action: run a controlled failover when the cluster is quiet. Until it works, schedule the landing with an explicit outage mode: stale publish remains, checkpoint does not move, the mart is late and labeled late.

Symptom: Object-store restatement time is about twice the HDFS test, and one date is half updated.

Likely cause: rename-style commit implemented as copy, plus a reader that does not consult the checkpoint.

Investigation: committer class in the job configuration, object-store request metrics for copy versus multipart complete, and whether SQL uses the checkpoint.

Corrective action: switch to the documented committer, re-run the failed date, and block the reader on the checkpoint. Keep the HDFS test for logic. Add an object-store failure test for visibility.

Symptom: A prohibited field appeared in the published Parquet and is now in three replicas.

Likely cause: the check ran after commit, or a schema-on-read job ingested partner files directly into the published tree.

Investigation: job logs for the refusal path, and whether the compact job is actually the only writer.

Corrective action: replace the date with a clean run, confirm replicas of the bad blocks are gone after the delete policy you trust, and move the check before commit. Deleting a path on HDFS removes namespace entries and schedules block deletion. Confirm the deletion with the cluster's trash setting. Trash can keep the bad file recoverable, which is good for accidents and bad if you believed the delete was immediate.

A design review asks: Where is the namespace, and is it highly available in a failover you have seen? What is the block size, the replication, and the file count of one peak date? Where is the checkpoint relative to the dataset root? What is the commit operation on this filesystem, and what does a reader see if it fails halfway? Who can read the prefix? What happens to a DataNode during the run? Which inherited MapReduce jobs still publish, and do they fail closed on collisions? The review does not start by adding a second cluster.

Interview Questions and Answers

Q1. What does the NameNode store, and what does it refuse to store?

It stores the namespace: paths, permissions, replication, and the block list for each file, plus which DataNodes reported those blocks. The durable form is the fsimage and the edit log. It does not store the contents of the progress files. Clients read and write bytes on DataNodes. If I send learners' events through the NameNode as a data path, I have misunderstood the design and I will take the namespace down. The operational consequence is that NameNode loss hides every path even when the disks are fine, which is why HA is a separate question from replication.

Q2. Why is a 10 MB file not a 128 MB block on disk, and why is it still a problem?

The block is a maximum replication unit. The last block, and a small file's only block, occupy the bytes they need. Disk capacity is not rounded up to 128 MB. Namespace memory is still charged for the file and the block. Fifty million 10 KB files are a NameNode problem and a listing problem, not a capacity-rounding problem. I compact them into Parquet files that approach the block size so each inode represents a useful sequential read.

Q3. How does a client write one block of the progress publish?

The client asks the NameNode to create the file and allocate a block. The NameNode returns a pipeline of DataNodes, placed with the rack policy if topology exists. The client streams packets to the first DataNode, which forwards downstream. Acks return upstream. When the client closes, the NameNode finalizes the block list. Until close and until the commit places that file in the published date, the SQL job must not read it. Replication happens as part of the pipeline, not as a later backup job, though re-replication is a later job if a replica is lost.

Q4. What is the default replica placement trying to survive?

Loss of a single node and, when racks are known, loss of a single rack for a block that has three replicas: one placement is local to the writer when possible, and the other two sit on a different rack, on two nodes. I verify topology instead of assuming the script was installed. Replication copies whatever bytes I committed, so it survives hardware loss and faithfully preserves a wrong file. The contract check is what survives a wrong file.

Q5. How does a standby NameNode differ from a Secondary NameNode?

The standby in an HA pair applies the same edits, typically from a quorum of JournalNodes, and can become active when the failover controller fences the old active. The Secondary NameNode creates periodic checkpoints of fsimage so edit logs stay manageable. It is not waiting to serve clients. I look for JournalNodes and a successful failover drill before I tell TechLearningPro the raw prefix is highly available. A checkpoint node plus hope is a single NameNode.

Q6. When would you consider erasure coding for progress data?

For cold date partitions that are rarely read, large, and already compacted, where the storage multiplier of three replicas dominates cost and the CPU cost of reconstruction is acceptable. I leave the hot date we replace each morning replicated, because we read and rewrite it under a bound. Erasure coding is a cluster policy with named layouts such as Reed-Solomon stripes in Hadoop 3, not a Spark write option I flip in a job. Tiny files remain the wrong input to either policy.

Q7. What does YARN decide, and what does it not decide?

YARN decides which NodeManagers run which containers for which application's ApplicationMaster, inside queue capacities. It does not understand event_id, the progress schema, or the rule that two jobs must not commit the same event_date. The landing's single-writer rule lives in the scheduler and the checkpoint. MapReduce and Spark are applications on top. I use a queue so the quarter restatement cannot take every container, and I do not expect the queue to validate JSON.

Q8. Why can the same Spark overwrite be safe on HDFS and unsafe on an object store?

HDFS rename of the attempt directory into the published directory is a namespace operation. The Hadoop S3A layer's historical rename is a copy and delete of objects, which can fail in the middle and leave both generations visible to a list. Hadoop documents committers so the object becomes visible when complete. I test the failure on the store I will use. I also keep a checkpoint the SQL job requires, because a glob does not honor _SUCCESS. The PySpark transformation can be identical. The commit configuration cannot be an unexamined copy.

Q9. Where does the checkpoint live, and why not beside the Parquet files?

Outside the dataset root, under a prefix the landing role controls, keyed by product and input name, written after the date commit succeeds. Inside the date directory it can be deleted with the partition, picked up by a glob, or parsed as if it were part of the table. _SUCCESS stays where the committer writes it and means the committer finished. The checkpoint means this input hash produced this publish. Those are different sentences, and the reader of completion uses the second.

Q10. A MapReduce job keeps the latest hash for each event_id. What do you change?

I change it to fail the run when two canonical hashes appear for one event_id, and to collapse identical hashes to one. That is the contract from the foundations lesson and from the Python landing. "Latest" depends on file order or on a timestamp the exporter can rewrite, and it publishes one of the payloads without an alarm. Counters surface the conflict. The driver checks them before rename. I would rather miss the morning bound than ship a second completion rate.

Q11. How do you explain a missing block to someone who set replication to three?

I check the actual replication of the path, the topology, and fsck output for under-replicated or missing blocks. The usual causes are an override to fewer replicas, all replicas placed in one failed domain because rack awareness was off, or more failures than the remaining replicas during a long outage before re-replication completed. I do not serve a partial day to the mart while blocks are missing. I restore or I mark the date stale.

Q12. What is the relationship between an HDFS block and a Spark partition?

A block is a storage and replication unit. A Spark partition is a task's input slice. On HDFS those slices are usually block-aligned, and both defaults are 128 MB, so they often coincide. They are still different settings: dfs.blocksize and spark.sql.files.maxPartitionBytes. A Parquet row group is a third unit inside the file. I set file sizes so a task reads a sensible amount of one date, and I do not tell a reviewer that "we set the partition to 128 MB" without saying which of the three I changed.

Hands-On Exercises

Exercise: Layout and failure memo for the progress prefix

Problem statement. Specify the HDFS or object-store layout for TechLearningPro's published progress events, with numbers and a failure story. You do not need a cluster if you label commands you could not run.

Tasks.

1. Draw the dataset root, one date partition, the checkpoint prefix, and the roles that may read each.
2. Compute blocks and replicas for a columnar layout and for a one-file-per-event layout, using a stated byte size and a 128 MB block. Label estimates.
3. State replication or the object-store durability you are relying on, and what a single disk loss does.
4. Describe commit visibility on the filesystem you chose. If it is an object store, name the committer family you would verify in the docs, without inventing a class.
5. Place YARN or another resource layer in one paragraph: what it schedules, and what the single-writer rule still has to cover.
6. Read the collision reducer behavior: identical hashes versus two hashes. Say what the driver does before rename.
7. Explain Secondary NameNode versus standby in your own words, and what you would look for on a cluster you inherited.
8. Apply the six-part lens. Include NameNode memory under the bad layout as a cost and reliability item.

Expected deliverables. The layout, the arithmetic, the commit-and-checkpoint note, and the lens paragraphs.

Evaluation rubric.

You pass when a reviewer can point at one path and say who reads it, how many replicas a peak day creates in your columnar layout, and what remains visible if the commit fails halfway.

Lesson Summary

HDFS keeps a namespace on the NameNode and bytes on DataNodes, in blocks replicated across machines and, when topology is real, across racks. A standby NameNode with a journal is high availability. A Secondary NameNode is a checkpoint helper. The progress publish uses a date directory of reasonably large Parquet files so the namespace stays small, a commit that hides partial attempts, and a checkpoint outside the dataset written afterward. YARN schedules containers for whatever application you run. MapReduce is the inherited shape of a shuffle, not the landing you are building. An object store needs its own committer because rename may be a copy. Replication makes hardware loss survivable and faithfully stores whatever you committed, so the contract check still happens first.

Apache Spark Architecture is the next lesson: the driver, the executors, and the jobs that read this layout. The filesystem will still be here when that lesson talks about tasks.

Further Reading

• HDFS architecture guide — blocks, replicas, NameNode, DataNode, and the write path. This is the source for the model in this lesson.
• HDFS high availability — Quorum Journal Manager and standby NameNodes. Use it to check the failover story of any cluster you inherit.
• HDFS erasure coding — Hadoop 3 policies and the storage-versus-CPU trade. Read it before enabling a policy on cold dates.
• YARN overview — ResourceManager, NodeManager, ApplicationMaster.
• Hadoop AWS committers — why S3A commit is not an HDFS rename. Confirm the keys against the Hadoop version you run.
• Apache Spark cluster overview — the next lesson's map, including YARN as one cluster manager.
• Continue this course: Apache Spark Architecture, then Spark DataFrames and Spark SQL. The Python landing contract remains Building Batch Data Pipelines with Python.