Distributed Storage - HDFS & Object Stores

The exponential growth of global data generation—spanning transactional event logs, telemetry streams, high-resolution sensor metrics, and large language model training corpora—fundamentally outstripped the scaling limits of single-node storage architectures and symmetric multiprocessing (SMP) enterprise storage area networks (SANs/NAS). Traditional relational databases and POSIX file systems were engineered around assumptions of localized hardware reliability, vertical scaling, hardware RAID arrays, and strict POSIX filesystem semantics (such as directory-level locking, random file mutations, and atomic renames).

In distributed big data systems, these assumptions are inverted. Distributed storage architectures operate on commoditized, heterogeneous hardware where component failures (hard drive bad sectors, top-of-rack switch partitions, node kernel panics) are statistically guaranteed continuous events. Pioneered by Google's landmark Google File System (GFS) paper (Ghemawat et al., 2003) and open-sourced via the Hadoop Distributed File System (HDFS), distributed storage divorces physical disk topology from logical access, transitioning from fragile hardware redundancy to software-managed replication, Reed-Solomon erasure coding, and cloud-native object storage primitives.

1. Theoretical & Architectural Foundations#

The HDFS Master-Worker Architecture#

HDFS implements a asymmetric master/worker distributed filesystem optimized for streaming access patterns across multi-gigabyte or multi-terabyte files ("write once, read many times").

+-----------------------------------------------------------------------------------+
|                        HDFS DISTRIBUTED STORAGE CLUSTER                           |
+-----------------------------------------------------------------------------------+
|                            ACTIVE NAMENODE (MASTER)                               |
|   - In-Memory Metadata Namespace: File-to-Block Inode Map, Permissions, Generation|
|   - EditLog (WAL on JournalNodes) & FSImage Checkpoints (Secondary NameNode)      |
|   - Block Placement Engine & Lease Manager (Write Locks)                          |
+------------------------------------------+----------------------------------------+
                                           | Heartbeats (3s) & Block Reports (6h)
         +---------------------------------+---------------------------------+
         |                                 |                                 |
         v                                 v                                 v
+------------------+             +-------------------+             +------------------+
| DATANODE 1       |             | DATANODE 2        |             | DATANODE 3       |
| (Rack 1)         |             | (Rack 2)          |             | (Rack 2)         |
+------------------+             +-------------------+             +------------------+
| - Blk_101 (128MB)|             | - Blk_101 (Repl 2)|             | - Blk_101 (Repl 3|
| - Blk_102 (128MB)|             | - Blk_104 (128MB) |             | - Blk_105 (128MB)|
| - Checksums (CRC)|             | - Checksums (CRC) |             | - Checksums (CRC)|
+------------------+             +-------------------+             +------------------+

1. NameNode (Metadata Master):

* Maintains the filesystem directory tree and the mapping of file path inodes to ordered sequences of logical block IDs entirely in memory (RAM).

Persists namespace state durably using two constructs: FSImage (a point-in-time serialized snapshot of the entire namespace) and EditLog* (a high-throughput Write-Ahead Log recording every transaction: file creation, block allocation, deletion, permission mutations).

High Availability Architecture*: Production deployments eliminate the single point of failure (SPOF) using an Active/Standby NameNode pair coordinated by a Quorum Journal Manager (QJM)—a Raft/Paxos-like consensus cluster of at least 3 JournalNodes. The Standby NameNode continuously tails edits from the QJM, keeping its in-memory state synchronized for instantaneous failover orchestrated via ZooKeeper (ZKFailoverController).

1. DataNodes (Block Storage Workers):

* Store physical raw block data files and accompanying .meta checksum files (CRC-32C) directly on local underlying Linux filesystems (ext4/XFS).

* Maintain zero metadata regarding file path hierarchies; a DataNode knows only block IDs (e.g., blk_1073741825).

Emit periodic heartbeats (default every 3 seconds) to signal node liveness, and comprehensive Block Reports* (default every 6 hours) informing the NameNode of all healthy blocks residing on their local disks.

Block Size Mechanics and the Small Files Pathology#

Unlike traditional OS filesystems that allocate space in 4 KB clusters, HDFS operates with large default block sizes (128 MB or 256 MB).

Architectural Rationale for Large Blocks:

• Amortizing Disk Seek Latency: Modern enterprise mechanical hard disk drives (HDDs) have an average seek time of t_{seek} ≈ 8 ms to 10 ms and a sustained sequential transfer rate of R_{seq} ≈ 150 MB/s. By allocating 128 MB blocks, transfer time is:

t_{transfer} = (128 MB / 150 MB/s) ≈ 0.853 seconds = 853 ms

The seek overhead is:

(t_seek / (t_seek + t_transfer)) = (10 ms / 10 ms + 853 ms) = (10 / 863) ≈ 1.16\%

Over 98.8\% of the physical drive operation is spent in continuous sequential streaming rather than high-latency arm actuator repositioning.

The Small Files Problem:

Every file, directory, and distinct block in HDFS occupies an independent metadata object in the NameNode's JVM heap, requiring approximately 150 bytes of memory.

If a data platform stores 100 million 10 KB files:

• Physical disk storage consumed: 100× 10^6 × 10 KB × 3 replicas = 3 TB (trivially small).
• NameNode JVM heap consumed: 100× 10^6 × 150 bytes ≈ 15 GB of raw resident pointer references.
• Consequence: The NameNode exhausts physical RAM, garbage collection pause times explode to multiple minutes (causing cluster-wide heartbeat timeouts and false failovers), and distributed query engines (Spark/MapReduce) spend 90% of job runtime initiating millions of separate task JVMs to process microscopic data partitions.

HDFS Replica Placement Policy & Network Rack Awareness#

To survive catastrophic rack-level hardware failure (such as a top-of-rack switch failure or power distribution unit outage) while minimizing expensive cross-rack network core bandwidth, HDFS enforces the Rack Awareness Placement Policy for standard 3-way replication (R = 3):

+-----------------------------------------------------------------------------------+
|                        RACK-AWARE REPLICA PLACEMENT                               |
+-----------------------------------------------------------------------------------+
|  CLIENT WRITE PIPELINE:                                                           |
|                                                                                   |
|  [Client Node]                                                                    |
|        |                                                                          |
|        v (Intra-Rack Streaming)                                                   |
|  [DataNode 1 (Local Node / Rack A)]                                               |
|        |                                                                          |
|        v (Cross-Rack Core Switch Transit)                                         |
|  [DataNode 2 (Remote Node / Rack B)]                                              |
|        |                                                                          |
|        v (Intra-Rack Top-of-Rack Switch Transit)                                  |
|  [DataNode 3 (Remote Node 2 / Rack B)]                                            |
+-----------------------------------------------------------------------------------+

1. Replica 1: Placed on the local node where the client is executing. If the client runs outside the cluster, a DataNode is selected at random within a rack chosen for low utilization.
2. Replica 2: Placed on a DataNode located in a completely different, remote rack (e.g., Rack B), selected at random. This guarantees that if the entire primary rack (Rack A) suffers a power failure, the block remains accessible.
3. Replica 3: Placed on a different DataNode within the same remote rack as Replica 2 (Rack B). This avoids saturating the cluster's high-tier aggregation and spine switches during the client write pipeline, since the data packets traverse the cross-rack link exactly once.
4. Subsequent Replicas (R > 3): Distributed randomly across remaining racks to prevent any single rack from hosting more than ⌊ (R - 1) / racks ⌋ + 1 replicas.

Erasure Coding: Reed-Solomon RS(k, m) vs. 3x Replication#

While standard 3x replication guarantees high read throughput and simple failure recovery, it incurs a massive 200\% storage overhead (a raw storage multiplier of 3.0×). In petabyte-scale data lakes, storing 10 PB of logical data requires 30 PB of physical disk space.

HDFS 3.x introduced Erasure Coding (EC), adapting Reed-Solomon linear error-correcting codes to distributed storage. In an RS(k, m) encoding scheme:

• A data block is partitioned into k data cells.
• The system computes m parity cells using Galois Field (GF(2^w)) matrix multiplication:

P = G × D

where D is the k × 1 vector of data cells, G is an m × k generator matrix derived from a Vandermonde or Cauchy matrix, and P is the m × 1 vector of parity cells.

• The resulting k + m cells are distributed across k + m distinct DataNodes and racks.
• Fault Tolerance: Any subset of k out of the k + m cells is mathematically sufficient to reconstruct the entire original dataset. The system tolerates the simultaneous loss of any m arbitrary DataNodes or racks without data loss.

Storage Multiplier Comparison:
Replication (3x):        [Data] [Data] [Data]            -> Overhead: 200% (3.0x multiplier)
Erasure Coding RS(6, 3): [D1][D2][D3][D4][D5][D6] [P1][P2][P3] -> Overhead:  50% (1.5x multiplier)
Erasure Coding RS(10, 4): 10 Data + 4 Parity            -> Overhead:  40% (1.4x multiplier)

Trade-offs of Erasure Coding:

• Advantage: RS(6, 3) cuts physical disk costs in half compared to 3x replication (saving millions in enterprise CAPEX) while tolerating 3 concurrent node failures (3x replication tolerates only 2).
• Disadvantage: Reconstructing a lost block requires reading from k surviving nodes across the network, generating heavy network I/O and consuming CPU cycles for matrix decoding. Consequently, EC is favored for cold and warm archival data lakes, while hot streaming files often utilize 3x replication.

2. Evolution to Cloud Object Stores: S3, GCS, and ADLS#

Architectural Divergence: Hierarchical File Systems vs. Object Stores#

In modern cloud computing, dedicated on-premises HDFS clusters have largely been superseded by Cloud Object Stores (Amazon Simple Storage Service - S3, Google Cloud Storage - GCS, Azure Data Lake Storage Gen2 - ADLS).

Separation of Compute and Storage#

The decoupling of compute and storage represents the defining architectural shift of modern cloud data engineering:

1. Independent Elasticity: In HDFS, adding storage required purchasing physical server chassis containing CPUs and RAM. If storage demands grew 10x faster than compute demands, expensive CPU cores sat idle. Decoupled object stores allow storage to scale to exabytes at low per-gigabyte costs while compute clusters spin up dynamically for analytics queries and shut down immediately upon completion.
2. Ephemerality: Compute clusters are disposable. If an Apache Spark cluster running on Kubernetes encounters kernel corruption, it is terminated and recreated in seconds without risking data corruption, as all persistent data resides durably in S3/GCS.

S3 Performance Engineering: Prefix Partitioning & Range Requests#

To maximize throughput when querying object stores, distributed engines utilize two fundamental mechanisms:

• Byte-Range HTTP Requests: S3 supports standard HTTP Range: bytes=start-end headers. Rather than downloading a 500 MB Parquet file to inspect its schema, an engine issues a lightweight HTTP GET requesting the last 64 KB of the object, reading the Parquet file metadata footer directly over the network.
• Request Parallelism and Key Prefix Partitioning: S3 scales request throughput horizontally across internal physical storage partitions based on object key prefixes. Each distinct partition supports at least 3,500 PUT/POST/DELETE requests per second and 5,500 GET/HEAD requests per second. Distributing files across distinct prefixes (e.g., s3://bucket/events/year=2026/month=10/...) prevents request throttling and allows distributed query engines to aggregate hundreds of gigabits per second of concurrent read bandwidth.

3. Modern Columnar File Formats: Apache Parquet vs. ORC vs. Avro#

Storing analytical data in raw text formats (CSV, JSON) imposes severe computational penalties: lack of typed schemas, high serialization/deserialization CPU overhead, absence of column compression, and the necessity of scanning 100% of bytes from disk even when a SQL query accesses only 2 out of 100 columns.

Row-Oriented Layout (CSV / JSON / Avro):
[Record 1: ID, Name, Timestamp, Value] [Record 2: ID, Name, Timestamp, Value] ...
-> Inefficient for Analytics: Reading "Value" requires reading every preceding field.

Column-Oriented Layout (Apache Parquet / ORC):
[Header]
  Row Group 1:
    Column Chunk (ID):        [ID_1, ID_2, ID_3, ... ID_n]
    Column Chunk (Name):      [Name_1, Name_2, ... Name_n]
    Column Chunk (Timestamp): [TS_1, TS_2, ... TS_n]
    Column Chunk (Value):     [Val_1, Val_2, ... Val_n]
[Footer: Schema, Row Group Metadata, Column Statistics: min, max, null_count]

The Anatomy of Apache Parquet#

Apache Parquet is an open-source, columnar storage format optimized for complex analytical queries (OLAP) on large datasets:

1. Row Groups: Logical horizontal partitions of data containing between 500,000 and 1,000,000 rows (typically sized between 128 MB and 512 MB uncompressed).
2. Column Chunks: Within each Row Group, data for each individual column is stored in contiguous byte arrays.
3. Pages: Column chunks are further subdivided into Pages (typically 1 MB), which represent the atomic unit of compression and encoding.
4. Footer Metadata: Placed at the very end of the file. Contains the schema definition, record count, and per-column statistics (minimum value, maximum value, null count) for every single row group.

Encoding and Compression Algorithms#

Parquet achieves dramatic compression ratios (typically 4× to 10× smaller than raw JSON) by combining domain-specific encodings with generic compression codecs (Snappy, Zstandard):

• Dictionary Encoding: Replaces repetitive string values with compact integer dictionary keys. If a country column contains 200 distinct country names across 10 million rows, strings are stored once in a dictionary page, and data pages contain lightweight 8-bit integers.
• Run-Length Encoding (RLE) & Bit-Packing: Compresses consecutive repeated values (e.g., [1, 1, 1, 1, 1, 2, 2] is encoded as (5 runs of 1), (2 runs of 2)).
• Delta Encoding: For monotonically increasing sequence IDs or timestamps, stores only the mathematical differences between consecutive values (Δ_i = v_i - v_{i-1}), drastically reducing numerical entropy.

Predicate Pushdown and Projection Pruning#

Columnar formats enable two fundamental query optimizations:

1. Projection Pruning: If a table has 150 columns but a query specifies SELECT user_id, amount FROM transactions, the query engine reads only the byte offsets corresponding to the user_id and amount column chunks. The remaining 148 columns are never fetched from disk or object storage, reducing physical I/O by > 95\%.
2. Predicate Pushdown: When evaluating WHERE amount > 10000, the query engine reads the Parquet footer first. It inspects the max_value statistic for the amount column in Row Group 1. If max_value = 4500, the engine mathematically proves that no row in Row Group 1 can satisfy the condition. The entire 128 MB Row Group is skipped completely without decompressing a single page.

4. Common Antipatterns & Production Pitfalls#

Antipattern 1: The "Small Files" Explosion on Object Stores#

Generating hundreds of thousands of tiny (< 1 MB) Parquet files due to excessive partitioning (e.g., partitioning by year/month/day/hour/minute/tenant_id):

• Pathology: Distributed engines spend hours issuing millions of HTTPS GET requests to S3. Network latency, SSL handshakes, and S3 listing limits completely dominate query execution time.
• Remediation: Compact files into target sizes of 128 MB to 512 MB using automated compaction utilities (e.g., Apache Iceberg/Delta Lake OPTIMIZE commands).

Antipattern 2: Blindly Using S3 as a POSIX Filesystem#

Attempting to treat object stores as traditional filesystems using tools like s3fs or relying on directory rename operations:

• Pathology: Renaming s3://bucket/data/old_dir/ to s3://bucket/data/new_dir/ containing 50 TB of data is not an atomic pointer swap; S3 must internally execute N separate copy operations followed by N delete operations, consuming hours and incurring massive API billing charges.
• Remediation: Use modern open table formats (Apache Iceberg, Delta Lake, Apache Hudi) that decouple table structure from physical file paths using ACID metadata manifests.

5. University-Grade Checkpoint Practice Problems#

Problem 1: Reed-Solomon Erasure Coding Storage & Bandwidth Math#

Context: An enterprise data lake currently stores S_{logical} = 12 PB (12,000 TB) of raw analytical data on an HDFS cluster using standard 3x replication.

The enterprise evaluates migrating the data to an Erasure Coding scheme using Reed-Solomon RS(6, 3) with cell size c = 1 MB and block group size B = 128 MB.

Financial & Hardware Metrics:

• Raw enterprise storage cost (amortized server hardware, power, cooling, disk maintenance): \0.015 per GB per month.
• 1 PB = 1,000 TB = 1,000,000 GB.

Task:

1. Calculate the total physical raw storage capacity consumed under 3x replication and under RS(6, 3) erasure coding.
2. Calculate the monthly financial savings achieved by converting the 12 PB data lake from 3x replication to RS(6, 3).
3. Calculate the maximum number of concurrent DataNode failures the cluster can tolerate under both schemes without suffering permanent data loss.
4. Suppose a DataNode hosting 40 TB of RS(6, 3) data blocks suffers a catastrophic hard drive controller failure. Calculate the exact total network traffic (in TB) that must be transmitted across the cluster spine network to reconstruct the lost 40 TB of data onto a spare node.

Solution:

1. Physical Storage Capacity Comparison:

* Under 3x Replication:

S_{physical, 3x} = 3 × S_{logical} = 3 × 12 PB = 36 PB = 36,000,000 GB

* Under Reed-Solomon RS(k, m) = RS(6, 3):

* Data cells k = 6.

* Parity cells m = 3.

* Storage expansion multiplier:

M_{RS} = (k + m / k) = (6 + 3 / 6) = (9 / 6) = 1.50 (50\% storage overhead)

* Total physical capacity:

S_{physical, RS} = 1.5 × 12 PB = 18 PB = 18,000,000 GB

* Physical storage saved: 36 PB - 18 PB = 18 PB of raw disk capacity.

1. Monthly Financial Savings:

* Storage capacity freed: 18,000,000 GB.

* Monthly savings:

Savings = 18,000,000 GB × \0.015 / GB / month = \270 USD,000 per month

* Annualized cost reduction:

Annual Savings = \270,000 × 12 = \3,240,000 per year

1. Fault Tolerance Comparison:

* 3x Replication: Retains 3 total copies. If 2 nodes fail, 1 copy survives. If 3 nodes fail, data is lost. Maximum tolerable concurrent node failures: F_{3x} = 2 nodes.

* RS(6, 3) Erasure Coding: Generates k + m = 9 fragments across 9 distinct nodes/racks. Any k = 6 surviving fragments can reconstruct the original block via matrix inversion. Maximum tolerable concurrent node failures: F_{RS} = m = 3 nodes.

Result*: RS(6, 3) delivers higher fault tolerance (3 node failures vs 2) while cutting physical storage consumption in half.

1. Network Reconstruction Bandwidth:

* Under RS(k, m), reconstructing any 1 lost data fragment requires reading exactly k surviving fragments (either data or parity) from the remaining nodes.

* To reconstruct D_{lost} = 40 TB of fragments, the cluster must stream:

Reconstruction Traffic = k × D_{lost} = 6 × 40 TB = 240 TB

Architectural Insight*: Under 3x replication, restoring a failed 40 TB node requires transferring only 1 × 40 TB = 40 TB from a mirror node. Under RS(6, 3), network reconstruction requires 6× more data transit (240 TB), representing the classical trade-off between disk storage savings and network/CPU reconstruction overhead.

Problem 2: NameNode JVM Memory Sizing & Small Files Pathology#

Context: An HDFS deployment operates an Active NameNode configured with a JVM maximum heap limit of H_{total} = 64 GB.

NameNode JVM Architecture Constants:

• 25\% of the heap is reserved for JVM runtime operational overhead, thread stacks, RPC handler buffers, and garbage collection working memory. The maximum heap allocated to namespace metadata is H_{meta} = 0.75 × 64 GB = 48 GB.
• The NameNode requires an average of M_{obj} = 150 bytes of heap memory per metadata entity (an entity is defined as an individual file, a directory inode, or a distinct block replica map entry).
• Default HDFS block size is B = 128 MB.

A poorly designed IoT ingestion pipeline writes raw sensor telemetry as millions of individual small JSON files, each with an average file size of S_{file} = 16 KB (0.016 MB).

Task:

1. Calculate the absolute maximum number of metadata entities N_{max} that can be stored in the NameNode's 48 GB metadata heap before risking fatal JVM OutOfMemory errors.
2. If the IoT pipeline generates 120 million individual 16 KB files (each occupying 1 block), calculate the total NameNode heap memory consumed by this dataset alone, and determine if it exceeds heap capacity.
3. Calculate the total logical data volume of these 120 million files, and compare it to the theoretical maximum storage capacity the NameNode could support if files were sized at the full 128 MB block size.
4. Formulate the architectural remediation strategy using Hadoop Archive (HAR) or Parquet compaction.

Solution:

1. Maximum Allowable Metadata Entities N_{max}:

* Available metadata heap:

H_{meta} = 48 GB = 48 × 1,024^3 bytes = 51,539,607,552 bytes

* Memory per entity: M_{obj} = 150 bytes.

* Maximum entity count:

N_{max} = (51,539,607,552 bytes / 150 bytes/entity) ≈ 343,597,383 entities (≈ 343.6 million)

1. Heap Memory Consumed by IoT Small Files:

* Each file requires: 1 File Inode + 1 Block Inode = 2 metadata objects in the NameNode heap.

* For 120 million files:

N_{IoT} = 120,000,000 × 2 = 240,000,000 metadata objects

* Heap memory consumed:

H_{IoT} = 240,000,000 × 150 bytes = 36,000,000,000 bytes ≈ 33.53 GB

* Fraction of available metadata heap consumed:

(33.53 GB / 48.0 GB) × 100\% ≈ 69.85\%

Evaluation*: A single ingestion pipeline consumes nearly 70% of the entire enterprise NameNode's memory capacity.

1. Logical Volume vs. Theoretical Capacity:

* Logical data volume of the 120 million 16 KB files:

V_{IoT} = 120,000,000 × 16 KB = 1,920,000,000 KB = 1.92 TB

* Now, calculate the logical data volume if those same 240 million metadata objects were utilized for fully packed 128 MB files (where 1 file = 1 block):

V_{optimal} = 120,000,000 × 128 MB = 15,360,000 GB = 15.36 PB

* Ratio of capacity efficiency:

(V_optimal / V_IoT) = (15.36 PB / 0.00192 PB) = 8,000×

Pathology Summary*: The IoT pipeline wastes 99.9875\% of the NameNode's potential storage capacity management potential, exhausting server memory to store a meager 1.92 TB of data.

1. Remediation Architecture:

Streaming Ingestion Compactor*: Deploy an intermediate buffering service (Kafka + Spark Structured Streaming / Apache Flink) that aggregates incoming 16 KB sensor payloads in memory and flushes them to HDFS as consolidated 128 MB Parquet files once every 15 minutes.

Result*: Reduces the 120 million files down to 1.92 TB / 128 MB = 15,000 files, slashing NameNode metadata objects from 240,000,000 to 30,000, reclaiming over 33 GB of RAM.

Problem 3: Parquet Predicate Pushdown and Projection Pruning I/O Analysis#

Context: A large analytical dataset comprising 10 billion records of financial market transactions is stored in a cloud object store.

Dataset Schema & Column Profile:

• Total columns: C_{total} = 40 columns.
• Average uncompressed record width across all 40 columns: 200 bytes/row.
• Total uncompressed raw data volume:

V_{raw} = 10 × 10^9 × 200 bytes = 2,000,000,000,000 bytes = 2.0 TB

Storage Configurations under Comparison:

• Format A (Uncompressed CSV): Full row-oriented plaintext storage.
• Format B (Apache Parquet with Snappy Compression):

* Columnar format partitioned into 4,000 Row Groups (each containing 2,500,000 rows).

* Average columnar compression ratio across all columns: 4:1 (compressed size is 25\% of uncompressed).

* The dataset includes column transaction_id (8 bytes uncompressed), timestamp (8 bytes uncompressed), trade_symbol (8 bytes uncompressed), and trade_price (8 bytes uncompressed).

* The remaining 36 columns represent secondary dimensions totaling 168 bytes/row uncompressed.

Query Execution:

A business intelligence analyst executes the query:

SELECT trade_symbol, AVG(trade_price) 
FROM market_transactions 
WHERE timestamp BETWEEN '2026-10-31 09:00:00' AND '2026-10-31 10:00:00'
GROUP BY trade_symbol;

Telemetry reveals that the 1-hour temporal filter matches records located in exactly 120 out of the 4,000 total Row Groups.

Task:

1. Calculate the total data volume (in GB) scanned across the network from the object store when executing the query against Format A (CSV).
2. Calculate the total data volume scanned when executing against Format B (Parquet) taking advantage of both Projection Pruning and Predicate Pushdown.
3. Compute the absolute I/O reduction (in GB) and percentage scan efficiency achieved by Parquet.

Solution:

1. Format A (CSV) Scan Volume:

* In a row-oriented CSV file, columns cannot be read independently, nor can row groups be skipped without scanning delimiters.

* To evaluate the WHERE clause and aggregate trade_price, the engine must execute a full table scan over the entire uncompressed dataset:

V_{scan, A} = 2.0 TB = 2,000 GB

1. Format B (Parquet) Scan Volume:

Step 1: Parquet Footer Metadata Read*:

* The query engine issues HTTP byte-range requests to read the file footer metadata for all row groups. Footers average 32 KB per row group:

V_{footer} = 4,000 × 32 KB = 128,000 KB ≈ 0.128 GB

Step 2: Predicate Pushdown Evaluation*:

* The query engine inspects the min/max statistics for column timestamp in the footers.

* 3,880 out of 4,000 row groups have timestamps entirely outside the query window. These 3,880 row groups are pruned completely (0 bytes scanned).

* Only N_{active} = 120 row groups require column chunk retrieval.

Step 3: Projection Pruning*:

* The query only references three columns: trade_symbol (8 bytes), trade_price (8 bytes), and timestamp (8 bytes).

* Uncompressed width of requested columns:

W_{proj} = 8 + 8 + 8 = 24 bytes/row

* Uncompressed data in the 120 active row groups (120 × 2,500,000 rows = 300,000,000 rows):

V_{proj, uncompressed} = 300,000,000 × 24 bytes = 7,200,000,000 bytes = 7.2 GB

* Applying the 4:1 Snappy compression ratio:

V_{proj, compressed} = (7.2 GB / 4) = 1.8 GB

* Total Parquet scan volume:

V_{scan, B} = V_{footer} + V_{proj, compressed} = 0.128 GB + 1.8 GB ≈ 1.928 GB

1. Efficiency Comparison:

* Absolute network I/O saved:

Δ V = 2,000 GB - 1.928 GB = 1,998.072 GB saved

* Percentage reduction in scan I/O:

(1 - (1.928 / 2,000)) × 100\% = (1 - 0.000964) × 100\% ≈ 99.90\%

* Parquet eliminated 99.9\% of physical data transfer, reducing query run time from tens of minutes down to less than two seconds.

Problem 4: HDFS Network Cross-Rack Ingress Replication Bandwidth#

Context: An enterprise HDFS cluster comprises 20 racks connected via top-of-rack (ToR) switches to a spine aggregation network switch.

An analytics ingestion pipeline streams raw data into the cluster at a sustained client write rate of:

R_{client} = 2.4 Gbps (300 MB/s)

The cluster enforces standard 3x Rack-Aware Replication:

• The client runs on a node located in Rack 1.
• Replica 1 is written to DataNode A on Rack 1.
• Replica 2 is written to DataNode B on Rack 2.
• Replica 3 is written to DataNode C on Rack 2.

In HDFS client write pipelines, data is streamed in a daisy-chain pipeline:

Client → DataNode A → DataNode B → DataNode C

Task:

1. Trace the network flow of data packets through the top-of-rack switches and spine network.
2. Calculate the sustained bandwidth consumed across the Rack 1 ToR uplink/downlink switch.
3. Calculate the sustained bandwidth consumed across the Rack 2 ToR uplink/downlink switch.
4. Calculate the sustained bandwidth traversing the core spine network link.
5. Compare this cross-rack spine bandwidth consumption with a naive replication policy that places each of the 3 replicas on 3 completely different racks (Rack 1, Rack 2, Rack 3).

Solution:

1. Pipeline Network Flow Analysis:

* Segment 1 (Client → DataNode A): Both nodes reside in Rack 1. Packets traverse only the internal Rack 1 ToR switch backplane. (Zero spine bandwidth).

* Segment 2 (DataNode A → DataNode B): DataNode A is in Rack 1; DataNode B is in Rack 2. Packets must ascend Rack 1 ToR uplink, traverse the core spine network switch, and descend Rack 2 ToR downlink.

* Segment 3 (DataNode B → DataNode C): Both nodes reside in Rack 2. Packets traverse only the internal Rack 2 ToR switch backplane. (Zero spine bandwidth).

1. Rack 1 Switch Bandwidth Consumption:

* Ingress from client into DataNode A: 2.4 Gbps (internal ToR switching).

* Egress from DataNode A to spine uplink: 2.4 Gbps.

* Total uplink egress from Rack 1:

B_{Rack 1, uplink} = 2.4 Gbps

1. Rack 2 Switch Bandwidth Consumption:

* Ingress from spine downlink into DataNode B: 2.4 Gbps.

* Forwarding from DataNode B to DataNode C: 2.4 Gbps (internal ToR switching).

* Total downlink ingress into Rack 2:

B_{Rack 2, downlink} = 2.4 Gbps

1. Core Spine Network Bandwidth Consumption:

* Only Segment 2 (DataNode A → DataNode B) traverses the core spine network.

* Sustained spine bandwidth:

B_{spine, rack-aware} = 2.4 Gbps

1. Comparison with 3-Distinct-Rack Policy:

* Under a naive policy placing Replica 1 on Rack 1, Replica 2 on Rack 2, and Replica 3 on Rack 3:

* Segment 1 (Client → DN1): Local to Rack 1.

* Segment 2 (DN1 → DN2): Cross-rack from Rack 1 to Rack 2 (2.4 Gbps over spine).

* Segment 3 (DN2 → DN3): Cross-rack from Rack 2 to Rack 3 (2.4 Gbps over spine).

* Total cross-rack spine bandwidth:

B_{spine, naive} = 2.4 Gbps + 2.4 Gbps = 4.8 Gbps

Architectural Conclusion*: HDFS Rack-Aware Placement cuts core spine switch bandwidth consumption exactly in half (2.4 Gbps vs 4.8 Gbps) while still providing total immunity to single-rack physical disaster.