Hadoop & HDFS Reference

Big data fundamentals, the Hadoop Distributed File System, the HDFS CLI, and Hive query tuning.

1. What is Big Data

Big data describes data whose volume, velocity and variety exceed what traditional databases and tools can capture, store and process. As internet, mobile and social usage grew, storing and managing that data became the central engineering challenge — not just analyzing it.

The 5 V's

Types of data

• Structured — fixed schema, fits neatly into tables (roughly 20% of data). Easy to store and query in relational databases.
• Semi-structured — no fixed schema but some structure, e.g. XML or JSON. Needs processing before it fits a table.
• Unstructured — no structure at all: images, video, free text (roughly 80% of data).

2. Apache Hadoop

Apache Hadoop (2006) is an open-source framework for distributed storage and processing of big data across clusters of commodity hardware. It assumes hardware failure is common and handles it automatically, making it highly scalable and fault tolerant. Hadoop splits files into large blocks, distributes them across cluster nodes, and processes them in parallel using a master–slave architecture.

3. HDFS Architecture

HDFS follows a master–slave design. The NameNode (master) holds filesystem metadata — the directory tree and which blocks live on which nodes — while DataNodes (slaves) store the actual block data and serve read/write requests. A Secondary NameNode periodically merges edit logs into the filesystem image; it is a checkpoint helper, not a hot standby.

• Block size — files are split into large blocks, default 128 MB, to keep seek overhead low relative to transfer time.
• Replication factor — each block is copied (default 3x) across DataNodes for fault tolerance.
• Rack awareness — replicas are spread across racks so a single rack failure doesn't lose all copies of a block.

4. Using the HDFS CLI

As a developer you typically connect to a gateway/client node over SSH, then interact with HDFS using hadoop fs or hdfs dfs — the two are aliases of each other for filesystem operations. hadoop has other subcommands beyond fs and covers HDFS/MapReduce interaction broadly; hdfs has other subcommands beyond dfs and covers administrative tasks (NameNode, Secondary NameNode, DataNode). As developers, day-to-day scope is limited to hdfs dfs / hadoop fs.

Anatomy of a command, using hdfs dfs -ls -l -S -r /public as an example:

hdfs        → main command, manages all HDFS components
  dfs       → sub-command, manages files in HDFS
    -ls     → file system command: list files
      -l -S -r  → control arguments (long listing, sort by size, reverse)
        /public → argument: the HDFS path to list

5. Command Reference

6. YARN Architecture

YARN separates resource management from job scheduling/monitoring, letting Hadoop run more than just MapReduce on one cluster.

• ResourceManager — one per cluster (with a standby for HA); tracks available resources and grants containers to applications.
• NodeManager — one per worker node; launches and monitors containers, reports node health and resource usage to the ResourceManager.
• ApplicationMaster — one per running application (a MapReduce job, a Spark app); negotiates containers from the ResourceManager and coordinates the application's own tasks inside them.
• Container — a bundle of resources (CPU, memory) on one node, the unit YARN actually allocates.

This is why Spark-on-YARN has a driver that runs as (or inside) the ApplicationMaster, and executors that run inside containers negotiated from the ResourceManager — the same YARN concepts underpin both classic MapReduce and modern Spark jobs.

7. MapReduce Phases

A MapReduce job processes data in four stages, each running in parallel across the cluster:

1. Map — each mapper reads one input split and emits key-value pairs.
2. Shuffle — the framework transfers each mapper's output to the reducer responsible for its key.
3. Sort — keys arriving at a reducer are sorted, so a reducer sees all values for a key together and in order.
4. Reduce — each reducer aggregates the values for its keys and writes final output, typically back to HDFS.

Shuffle is usually the most expensive phase — it moves data across the network — which is exactly why Spark's equivalent shuffle step is the first thing to tune (see §9 of the Spark reference doc).

8. HDFS Read/Write Path

Writing a file: the client asks the NameNode for a list of DataNodes to host the first block; it streams data to the first DataNode, which forwards it to the second, which forwards it to the third (a replication pipeline) — each DataNode acknowledges back up the pipeline once its copy is durable.

Reading a file: the client asks the NameNode for the block locations of a file; the NameNode returns DataNodes ordered by network proximity to the client, and the client reads each block directly from the nearest available replica — the NameNode itself never serves file data.

Safe mode — on startup, the NameNode enters safe mode: read-only, waiting for DataNodes to report in their block inventories before it considers the filesystem's replication state healthy enough to allow writes.

9. Hadoop Ecosystem

10. Hive Query Tuning

Limit optimization

A bare LIMIT still runs the entire query before truncating results — wasteful for ad-hoc CLI exploration. Enable the optimizer so Hive samples input instead of scanning it all:

SET hive.limit.optimize.enable = true;
SET hive.limit.row.max.size = 100000;
SET hive.limit.optimize.limit.file = 10;

Trade-off: with sampling enabled, a query that also needs a full scan (a JOIN or GROUP BY) risks never seeing all the relevant input.

Join strategies

• Left semi join — returns rows from the left table when a match exists on the right; cheaper than an inner join. Hive has no right semi join.
• Map-side join — broadcasts a small table to every mapper, avoiding a shuffle: SET hive.auto.convert.join=true; with the threshold hive.mapjoin.smalltable.filesize (default 25MB).
• Bucket map join — when both tables are bucketed on the join key and one bucket count is a multiple of the other, Hive joins matching buckets directly without fetching whole tables: SET hive.optimize.bucketmapjoin=true;
• Sort-merge bucket join — fastest, when bucketed tables share bucket count and are sorted on the join key: adds SET hive.optimize.bucketmapjoin.sortedmerge=true;
• Hive does not support non-equi joins, nor optimized right/full-outer joins.

Ordering data

Other levers

• Local mode — hive.exec.mode.local.auto=true runs small jobs on a single machine, skipping cluster overhead.
• Parallel execution — hive.exec.parallel=true runs independent stages of a query concurrently.
• Vectorization — processes 1024 rows per batch instead of one at a time: hive.vectorized.execution.enabled=true.
• JVM reuse — mapred.job.reuse.jvm.num.tasks amortizes JVM startup cost across many short tasks.
• Tez engine — hive.execution.engine=tez can be 100–300% faster than classic MapReduce.
• Single reducer for multi-GROUP BY — hive.multigroupby.singlereducer=true merges multiple GROUP BY operations into one job when their keys overlap.
• Dynamic partitioning — hive.exec.dynamic.partition.mode=strict requires at least one static partition to guard against accidentally overwriting everything; hive.exec.max.dynamic.partitions and its per-node variant cap how many partitions a job may create.

11. Replication & Fault Tolerance

HDFS splits files into blocks (128 MB by default, often 256 MB) and stores each block on several DataNodes. The NameNode keeps only metadata in memory: the directory tree, file → blocks, and block → DataNodes.

• Block size, not file size: a 10 MB file uses 10 MB of disk, but every file and block costs about 150 bytes of NameNode heap. That is the real reason small files hurt.
• Spark's default input split (spark.sql.files.maxPartitionBytes) is also 128 MB, so one block ≈ one task.

Replication factor and rack awareness

The default replication factor is 3 (dfs.replication, or per path with hdfs dfs -setrep). Default placement: replica 1 on the writer's node, replica 2 on a node in a different rack, replica 3 on another node in that same remote rack. Losing any single node or any single rack loses no data, at the cost of 3× storage. Hadoop 3 erasure coding (e.g. RS-6-3) gives similar durability at about 1.5× storage, mainly for cold data.

How each failure is handled

hdfs dfs -setrep -w 3 /data/txns                       # change replication, wait
hdfs dfs -stat "%o %r" /data/txns/part-0000            # block size, replication
hdfs fsck /data/txns -files -blocks -locations         # blocks and replica locations
hdfs dfsadmin -report                                  # live / dead DataNodes

12. How Hive Reads Files

Hive puts SQL on top of files. A table is just metadata in the Metastore (schema, location, format, partitions) laid over existing files: schema-on-read. Rows that do not fit the schema come back as NULLs instead of failing.

beeline / JDBC ──► HiveServer2 ──► Metastore (schema, location, format, partitions)
                       │
                       └──► Tez / Spark / MapReduce tasks ──► read HDFS blocks locally

CREATE EXTERNAL TABLE transactions (
  txn_id STRING, account_id STRING, channel STRING,
  amount DOUBLE, event_time TIMESTAMP)
ROW FORMAT DELIMITED FIELDS TERMINATED BY ','
STORED AS TEXTFILE
LOCATION '/data/txns/'
TBLPROPERTIES ("skip.header.line.count"="1");

SELECT channel, COUNT(*), SUM(amount) FROM transactions GROUP BY channel;

• Query flow: parse → Metastore lookup → block locations from the NameNode → plan (map, shuffle, reduce) → tasks run where the blocks are → a SerDe turns bytes into rows.
• Formats: text/CSV (slow), ORC (Hive native), Parquet (the standard), Avro (row-based with embedded schema).
• Spark uses the same tables via SparkSession.builder.enableHiveSupport().

Then vs now

13. Hive's Problems & How They Were Solved

2010  Hive on MapReduce
2013  Stinger: Tez, ORC, vectorization
2016  Hive 2 + LLAP
2018  Hive 3: ACID by default
2019+ Lakehouse table formats queried by Spark / Trino
Today the Hive Metastore lives on as a catalog; the Hive engine is mostly legacy.