Hadoop Overview

Apache Hadoop is a Java framework for storing data across a cluster and executing distributed applications near that data. Its original batch model combines three separate responsibilities:

1. HDFS stores one logical filesystem across machines.
2. YARN allocates cluster resources and launches application processes.
3. MapReduce divides a batch computation into map, shuffle, and reduce phases.

Hadoop Common supplies the configuration, filesystem abstractions, security support, native-library integration, and command launchers used by those projects. Separating the responsibilities matters because HDFS can store data for applications other than MapReduce, and YARN can schedule application frameworks other than MapReduce.

Hadoop Distribution and Components #

Hadoop is built with Maven and released as a set of JARs, scripts, XML configuration files, documentation, and optional native libraries. A binary distribution places the Java components under share/hadoop; its bin and sbin scripts assemble a class path and start client commands or daemons. Native libraries under lib/native provide selected performance-sensitive features, including compression and the libhdfs interface. Their availability depends on the operating system and architecture, so an installation should run hadoop checknative rather than assume that every optional native feature is loaded. Production Hadoop deployments are generally Linux-based.

The top-level projects have distinct roles:

• Hadoop Common contains the Configuration API, generic filesystem client, security utilities, shell launchers, and common services.
• HDFS implements the distributed filesystem, its metadata service, block storage service, administrative tools, and optional federation features.
• YARN provides cluster-wide resource management and a container execution protocol.
• MapReduce provides a batch programming model and a YARN application implementation, including its application master and job history service.

The release archive also contains optional connectors and tools. For example, the S3A connector brings AWS SDK dependencies, and the native directory may contain both shared objects and static archives. Archive size therefore does not predict the memory footprint of the NameNode, DataNode, ResourceManager, or application processes.

Use the Apache release page for the supported release line and verify an archive against its published signature and SHA-512 checksum. Source archives, binary archives, documentation archives, and architecture-specific binary archives serve different purposes; a file name alone is not a compatibility guarantee. A cluster’s client libraries, server daemons, and applications should use compatible Hadoop versions.

Hadoop Common and the Command Line #

The hadoop launcher is the common entry point. It establishes Hadoop’s environment and class path before dispatching a named command or a Java main class. hadoop classpath prints the effective application class path, which includes the configuration directory and the Common, HDFS, MapReduce, and YARN libraries in a standard distribution. Use that output to diagnose a manually launched Java process rather than copying paths from another installation.

The command namespace includes both application-facing and operational tools:

Some commands require optional modules or a server-side feature to be enabled. For example, federation balancing tools require the corresponding HDFS federation deployment. Treat hadoop --help from the installed distribution as the authoritative command list, because commands change between releases.

hadoop jar application.jar example.Main invokes an application’s main class through the launcher. MapReduce clients can also use yarn jar; both commands reach the Hadoop client environment in current distributions. Prefer the command documented by the deployed distribution and application framework, because a YARN application still needs the correct site configuration, credentials, and class path before it contacts the ResourceManager.

The hdfs, yarn, and mapred launchers expose project-specific commands. For example:

• hdfs dfs is the HDFS filesystem shell, while hadoop fs is the generic shell. With an HDFS default filesystem, their ordinary filesystem operations address the same service.
• hdfs dfsadmin, hdfs fsck, hdfs balancer, and hdfs haadmin support HDFS administration, health inspection, rebalancing, and high availability.
• yarn application, yarn node, yarn queue, yarn logs, and yarn top inspect applications, nodes, queues, aggregated logs, and live resource use.
• yarn rmadmin, yarn routeradmin, and yarn schedulerconf administer ResourceManager, federation-router, and scheduler features when enabled.
• mapred historyserver starts the MapReduce JobHistory Server; it is not the same process as YARN’s Timeline Service.

Daemon subcommands follow the form hdfs --daemon start namenode or yarn --daemon start nodemanager. They are useful for controlled local and small-cluster operations. A production service manager should provide the same environment, identity, configuration directory, log directory, and restart policy explicitly.

Configuration Layers and Precedence #

Every Hadoop process constructs one or more Configuration objects. Hadoop Common loads the core resources, and project code adds HDFS, YARN, or MapReduce resources. Site files contain deployment-specific values; default files are packaged with the corresponding module and document the defaults.

The standard files are:

core-default.xml, hdfs-default.xml, yarn-default.xml, and mapred-default.xml are class-path resources supplied by the installed version. Do not edit them: an upgrade replaces them, and a site file can override non-final defaults. Consult the matching version’s configuration reference when selecting a property, because names and defaults evolve.

The launch scripts obtain their configuration directory from HADOOP_CONF_DIR; when it is unset, a normal binary installation uses ${HADOOP_HOME}/etc/hadoop. The *-env.sh files control launcher-level settings such as Java selection, daemon heap options, log directories, and PID locations. Per-daemon variables such as HDFS_NAMENODE_OPTS, HDFS_DATANODE_OPTS, YARN_RESOURCEMANAGER_OPTS, and YARN_NODEMANAGER_OPTS add JVM options for the named process.

Command-line -Dname=value settings and JVM system properties can override configuration for a process, subject to Hadoop’s final-property rules. Use them for a controlled invocation or diagnostic, not as an undocumented replacement for a reviewed site configuration. A property in HDFS_NAMENODE_OPTS affects the NameNode process; it does not automatically configure DataNodes or YARN services.

A reliable configuration workflow is:

1. Put shared Common values in core-site.xml.
2. Put service-specific values in the corresponding site file.
3. Distribute the same reviewed configuration set to every host that runs the affected client or daemon.
4. Validate XML with hadoop conftest, inspect generated values with hdfs getconf or the relevant command, and restart only the services whose settings changed.
5. Record the effective values and service version during an incident, because a client-side -D override can produce behavior that does not match the files on disk.

Cluster Daemons #

A minimal HDFS and YARN cluster has a NameNode, at least one DataNode, a ResourceManager, and at least one NodeManager. MapReduce adds an ApplicationMaster per running job and can use a JobHistory Server after jobs finish. Other daemons are optional and depend on the deployment.

YARN also has optional services such as the Timeline Service, Shared Cache Manager, Router, Registry DNS server, and Global Policy Generator. Start them only when their feature has been configured. The historical yarn historyserver launcher for the application history service is deprecated in Hadoop releases that report it as such; use the documented Timeline Service launcher instead. It remains distinct from mapred historyserver.

The NameNode and DataNodes, or the ResourceManager and NodeManagers, may run on the same kernel during development. Their architecture permits it, but a shared host, power domain, or network path defeats the failure isolation that a distributed design intends to provide. Put NameNode and DataNode storage in separate failure domains in a production cluster, and reserve enough disk and network capacity for replication and recovery traffic.

HDFS Namespace and Block Storage #

HDFS presents directories and files to clients but stores a file as fixed-size blocks distributed across DataNodes. The NameNode owns the namespace: directories, file names, ownership, permissions, quotas, file-to-block mapping, and block-replica locations. It does not relay file data. A DataNode stores block replicas on local volumes and reports its state to the NameNode.

The separation determines the data path. To read a file, a client asks the NameNode for the block locations, chooses a nearby replica when possible, and streams bytes directly from that DataNode. To create a file, a client asks the NameNode to allocate a block and then writes packets through a DataNode pipeline. The first DataNode forwards packets to the next replica, and acknowledgements return through the pipeline to the client. The NameNode coordinates placement and metadata, while DataNodes carry the file contents.

The NameNode receives heartbeats and block reports from DataNodes. Missing heartbeats mark a node unavailable; reports allow the NameNode to detect under-replicated, over-replicated, or misplaced blocks and schedule corrective work. The default replication policy and rack awareness aim to preserve replicas across failure domains. An administrator should configure the network topology when racks or equivalent domains matter, because a replication factor alone does not describe placement quality.

HDFS metadata is stateful and critical. The NameNode persists an fsimage snapshot and an edits log. During startup it reconstructs the namespace from those records. Back up metadata using a tested procedure, monitor checkpoint age and edit-log growth, and test restoration with the same Hadoop version and storage layout. Copying a live namespace directory by hand is not a replacement for a coordinated backup and recovery procedure.

HDFS Addresses and Storage Directories #

fs.defaultFS identifies the default filesystem URI used by clients, for example:

<property>
  <name>fs.defaultFS</name>
  <value>hdfs://nn1.example.net:8020</value>
</property>

Clients without an explicit URI use that value. It also gives DataNode and other service configurations a common logical NameNode endpoint to reference. It does not by itself choose every NameNode listening address; HDFS server-side properties such as dfs.namenode.rpc-address define the NameNode RPC endpoint in configurations that need an explicit address.

The NameNode stores its metadata in dfs.namenode.name.dir. DataNodes store block replicas in dfs.datanode.data.dir, which may name several independent volumes. Configure every volume deliberately and monitor each volume. A DataNode can continue to use healthy volumes when a failed-volume policy allows it, but the lost replicas still require repair and the failed hardware requires investigation.

Running several DataNodes on one development host requires more than changing dfs.datanode.data.dir. Each instance needs separate storage directories and non-conflicting data-transfer, IPC, and HTTP addresses; each daemon should also have an unambiguous configuration and log/PID location. A NameNode instance likewise needs its own dfs.namenode.name.dir and RPC/HTTP addresses. Formatting creates storage identity, so never format a directory that contains a namespace or replicas that must be retained.

For a fresh, isolated development cluster, the order is:

1. Set fs.defaultFS and the HDFS site properties.
2. Set safe, empty NameNode and DataNode storage paths.
3. Format the NameNode once with hdfs namenode -format.
4. Start the NameNode and then DataNodes.
5. Confirm registration in the NameNode web UI and with hdfs dfsadmin reports before loading data.

The NameNode web interface is commonly exposed on port 9870 in Hadoop 3 defaults, but a deployment can change that address. Use the configured URL, not a copied port number, to inspect live DataNodes, capacity, block health, and filesystem information. Each DataNode also has an administrative web interface.

HDFS Clients and Operations #

HDFS is a distributed filesystem API and a shell interface, not a local POSIX filesystem. A FUSE-based integration can provide a mount in selected environments, but it does not make HDFS semantics or operational constraints identical to a local Unix filesystem. Use the HDFS client API or filesystem shell when application behavior depends on HDFS consistency, permissions, or atomic rename semantics.

The filesystem shell is command-oriented rather than an interactive REPL:

hdfs dfs -mkdir -p /data/input
hdfs dfs -put local-events.csv /data/input/
hdfs dfs -ls /data/input
hdfs dfs -cat /data/input/local-events.csv
hdfs dfs -rm -r /data/output

hadoop fs performs the same generic operations against the configured filesystem and can therefore address HDFS, local filesystems, or installed object-store connectors depending on the URI. Specify a URI when an operation must not depend on a caller’s fs.defaultFS.

Operational commands divide responsibility:

• Run hdfs dfsadmin -report to inspect DataNode capacity and service state.
• Run hdfs fsck /path to inspect block health; it reports a condition and does not replace repair planning.
• Run the HDFS balancer after changing capacity or placement, because new disks and nodes do not redistribute existing blocks automatically.
• Use hdfs cacheadmin, hdfs crypto, hdfs storagepolicies, hdfs ec, and snapshot commands only after configuring the corresponding HDFS feature.
• Use hdfs haadmin for controlled high-availability administration rather than attempting to promote a process by editing metadata files.

HDFS also supplies focused tools for specialized operating tasks. hdfs getconf exposes selected effective configuration values; hdfs jmxget queries JMX attributes; and offline image or edits viewers inspect metadata copies without making the live NameNode a general-purpose query service. hdfs diskbalancer addresses uneven use of volumes within a DataNode, whereas the cluster balancer moves replicas between DataNodes. hdfs mover enforces storage-policy placement after a policy change. Each tool changes a different layer of placement or metadata, so a report of free cluster capacity does not by itself identify the appropriate one.

Optional HDFS services extend the filesystem interface without changing the NameNode/DataNode data path. HttpFS exposes filesystem operations through an HTTP gateway, while HDFS NFS Gateway supports an NFS interface for compatible workloads. Router-based federation presents several namespace subclusters through routers and adds commands such as hdfs routeradmin, hdfs rbfbalance, and federation balancing tools. JournalNodes support NameNode HA by storing shared edits; they do not store ordinary file blocks. Deploy these services only when their additional availability, interoperability, or namespace requirements justify their operational cost.

Checkpointing and High Availability #

The SecondaryNameNode periodically obtains the NameNode image and edits, merges them into a new checkpoint, and returns the checkpoint for the active NameNode to use. Its name is historical: it is a checkpointing helper, not a hot spare and not a failover target. Losing the active NameNode is not solved by starting the SecondaryNameNode.

High availability uses separate active and standby NameNodes. The active NameNode processes namespace mutations; the standby applies the shared edits so it can take over when failover occurs. A typical deployment uses a quorum of JournalNodes for shared edit-log storage and ZooKeeper Failover Controllers for health monitoring and coordinated automatic failover. Fencing protects against two NameNodes becoming active at once.

HA therefore requires a complete design: distinct NameNode identities and addresses, shared-edit configuration, failover coordination, client failover configuration, fencing, and a tested recovery procedure. A replicated copy of dfs.namenode.name.dir does not provide those guarantees. Historical single-NameNode checkpointing and modern HA solve different operational problems and should not be described as interchangeable.

YARN Resource Management #

YARN separates resource management from a particular computation engine. Its primary actors are:

• The ResourceManager accepts applications, maintains cluster resource state, and uses a scheduler to allocate containers.
• A NodeManager runs on each worker host, reports node health and resource availability, localizes resources, launches containers, and monitors them.
• An ApplicationMaster is created for one application. It negotiates containers with the ResourceManager and coordinates that application’s work.
• A container is a resource allocation and process-execution boundary, identified by requested resources such as memory and virtual cores. It is not a container-image format.

The ResourceManager does not execute every task directly. It allocates an initial container for an application’s ApplicationMaster. After it starts, the ApplicationMaster requests additional containers, receives allocations, asks the relevant NodeManagers to launch processes, tracks progress, handles application-level retries, and releases resources. That delegation lets MapReduce, Spark, Tez, distributed services, and custom applications use a common cluster allocator while implementing different execution models.

The scheduler decides where and when resources are allocated according to its configured queues and policies. It does not prove that an application is correct, preserve an application’s intermediate state, or replace the ApplicationMaster’s retry logic. Capacity planning must account for NodeManager-advertised resources, YARN’s minimum and maximum allocation settings, queue limits, and the operating system resources reserved outside YARN.

Useful client operations include:

yarn node -list
yarn queue -status default
yarn application -list
yarn application -status application_1234567890123_0001
yarn logs -applicationId application_1234567890123_0001
yarn top

yarn logs needs log aggregation to be configured if logs are no longer on the NodeManager host. yarn schedulerconf changes scheduler configuration only when that administrative facility is enabled and authorized; a direct runtime change should follow the same review and rollback process as an XML configuration change.

Other YARN commands correspond to optional services rather than the core resource path. yarn timelinereader queries the Timeline Service reader when that service is deployed. yarn scmadmin administers the Shared Cache Manager, which can avoid localizing identical public resources repeatedly. yarn routeradmin applies to YARN federation routers, and yarn nodeattributes manages or queries node attributes used by placement policy. The globalpolicygenerator, router, sharedcachemanager, registrydns, and timelineserver daemon names therefore belong in a deployment runbook only after their matching service has been configured.

MapReduce as a YARN Application #

MapReduce is a batch model in which a mapper transforms input records into intermediate key-value pairs and reducers process groups of values for each key. It is not the storage layer and it is not YARN itself. On a modern Hadoop cluster, the MapReduce client submits an application to YARN, the MapReduce ApplicationMaster coordinates tasks, and task containers read and write data through HDFS or another configured filesystem.

A MapReduce job begins with a driver program that creates org.apache.hadoop.mapreduce.Job from a Configuration. The driver names the application JAR, mapper, reducer, input and output formats, map-output types, final-output types, input paths, output path, and reducer count. A framework default can be appropriate for a test, but an application should set semantics-critical choices explicitly because output types, partitioning, and paths determine correctness.

The submission lifecycle is:

1. The client validates the job and stages its JAR, configuration, split metadata, and other resources in the configured filesystem.
2. The client asks YARN’s ResourceManager to start the MapReduce ApplicationMaster.
3. The ApplicationMaster requests containers, preferring locations near input blocks where scheduler policy and capacity allow.
4. NodeManagers localize the submitted resources and launch map or reduce task JVMs in the allocated containers.
5. Tasks report status and counters to the ApplicationMaster. Failed attempts can be retried according to configuration; a task attempt is not the same as a logical task.
6. After output is committed, the ApplicationMaster reports final status to YARN. The JobHistory Server can subsequently serve completed MapReduce job details when configured.

YARN resource allocation and HDFS block locality reinforce each other but do not guarantee that every mapper runs on the DataNode holding its input. Queue pressure, node health, container sizes, and scheduler policy can require a remote read. Applications should benefit from locality without assuming it.

Building and Submitting a Job #

Use the hadoop-client dependency to compile a Java MapReduce client. Align its version with the deployed cluster’s supported client version rather than copying an old example version:

<dependency>
  <groupId>org.apache.hadoop</groupId>
  <artifactId>hadoop-client</artifactId>
  <version>${hadoop.version}</version>
  <scope>provided</scope>
</dependency>

provided keeps Hadoop’s large framework JARs out of an application JAR when the cluster supplies them. Applications that need third-party libraries must package or distribute them in a form available to task containers. Build the application with Maven or another Java build tool; invoking the JDK compiler through an old launcher example is not required for a normal build.

The MapReduce distribution also includes operational and compatibility tools. mapred job and its client APIs inspect submitted jobs, mapred historyserver serves completed-job records, and mapred pipes supports the historical Hadoop Pipes interface for native applications. Input sampling utilities can produce partitioning data for total-order workloads, while framework-upload tools can distribute a MapReduce framework archive when a cluster requires it. Those tools support the same application lifecycle; they do not bypass YARN’s application admission, container allocation, or HDFS staging rules.

A minimal word-count driver and mapper demonstrate the required declarations:

public final class WordCount {
  public static class CountMapper
      extends Mapper<LongWritable, Text, Text, IntWritable> {
    private static final IntWritable ONE = new IntWritable(1);
    private final Text word = new Text();

    @Override
    public void map(LongWritable offset, Text line, Context context)
        throws IOException, InterruptedException {
      for (String token : line.toString().split("\\s+")) {
        if (!token.isEmpty()) {
          word.set(token);
          context.write(word, ONE);
        }
      }
    }
  }

  public static class SumReducer
      extends Reducer<Text, IntWritable, Text, IntWritable> {
    @Override
    public void reduce(Text word, Iterable<IntWritable> counts,
                       Context context)
        throws IOException, InterruptedException {
      int sum = 0;
      for (IntWritable count : counts) {
        sum += count.get();
      }
      context.write(word, new IntWritable(sum));
    }
  }

  public static void main(String[] args) throws Exception {
    Configuration conf = new Configuration();
    Job job = Job.getInstance(conf, "word count");
    job.setJarByClass(WordCount.class);
    job.setMapperClass(CountMapper.class);
    job.setCombinerClass(SumReducer.class);
    job.setReducerClass(SumReducer.class);
    job.setOutputKeyClass(Text.class);
    job.setOutputValueClass(IntWritable.class);
    FileInputFormat.addInputPath(job, new Path(args[0]));
    FileOutputFormat.setOutputPath(job, new Path(args[1]));
    System.exit(job.waitForCompletion(true) ? 0 : 1);
  }
}

Prepare input, submit, and inspect output with separate commands:

hdfs dfs -mkdir -p /examples/wordcount/input
hdfs dfs -put input.txt /examples/wordcount/input/
hadoop jar wordcount.jar example.WordCount \
  /examples/wordcount/input /examples/wordcount/output
hdfs dfs -cat /examples/wordcount/output/part-r-00000

The output directory must not exist at submission time. MapReduce rejects an existing output path because replacing a previous result during a distributed job would make failed and retried attempts unsafe. Delete an output directory only after confirming that it is disposable, or choose a new run-specific path.

Input, Mapping, and Local Aggregation #

InputFormat maps configured input paths to InputSplit instances. A split is a unit of work metadata, not necessarily a physical HDFS block. File-based input formats often align splits with block boundaries to improve locality, but compression, file format, and custom input logic can change that relationship.

For each map task, a RecordReader turns one split into a sequence of input key-value records. TextInputFormat, for example, commonly supplies a byte offset as LongWritable and a line as Text. The framework creates a mapper instance and calls its lifecycle in this order:

1. setup(Context) runs once before records are processed.
2. map(KEYIN, VALUEIN, Context) runs once for each record.
3. cleanup(Context) runs once after the record reader is exhausted.

The generic mapper signature is:

public class Mapper<KEYIN, VALUEIN, KEYOUT, VALUEOUT> {
  protected void map(KEYIN key, VALUEIN value, Context context)
      throws IOException, InterruptedException {
    context.write((KEYOUT) key, (VALUEOUT) value);
  }
}

Applications override map to emit their intermediate representation. A map method must not retain mutable framework objects beyond the call unless it makes a safe copy, because Hadoop can reuse key and value objects while reading records.

Mapper output is buffered, partitioned, spilled, and sorted locally. A combiner can run as a local partial reduction before the shuffle, which can reduce network traffic. The framework may run a combiner zero, one, or many times, however, so its operation must preserve the result under repeated and partial application. A sum is safe for word count; an operation that depends on receiving every value exactly once is not. A combiner also does not replace the reducer’s final grouping semantics.

Partitioning, Shuffle, and Reduce #

Each map-output record is assigned to a reducer partition. The default partitioner hashes the map-output key, but Job#setPartitionerClass can select a custom partitioner. The driver sets the number of reduce tasks with Job#setNumReduceTasks; it is not derived from the number of map tasks. Setting zero reducers produces map-only output, while a positive reducer count creates one final output partition per reducer.

The shuffle transfers every mapper’s partition for reducer r to reducer r. Reducer tasks merge and sort those segments by key before invoking reduce. The default sort comparator compares the map-output key. A custom sort comparator can change ordering, and Job#setGroupingComparatorClass can define which adjacent sorted keys form one reduce group. Use custom comparators only with a partitioning strategy that sends all keys in a group to the same reducer.

The reducer lifecycle mirrors the mapper lifecycle:

1. setup(Context) runs once.
2. reduce(KEYOUT, Iterable<VALUEOUT>, Context) runs once for each grouped key after shuffle and sort.
3. cleanup(Context) runs once.

The iterable of values can stream from shuffle data. A reducer should process it during the call rather than retaining it for later use. The output format and output committer write task output to a work location and coordinate the job commit. The commit protocol is why output-path existence, speculative attempts, and filesystem rename behavior are operational concerns rather than mere command-line details.

Operating the Combined System #

Hadoop’s components form a deliberate dependency chain:

1. Hadoop Common loads configuration, supplies clients, and launches processes.
2. HDFS records where input and output blocks live, while clients transfer data directly to DataNodes.
3. YARN allocates the ApplicationMaster and task containers on available NodeManagers.
4. MapReduce uses those containers to run input reading, mapping, local sort, partitioning, shuffle, reducing, and output commit.

An operational investigation should follow the same chain. A job that cannot read input may have an HDFS path, permission, block-health, or DataNode problem before it has a MapReduce problem. A job that remains accepted may be waiting for an ApplicationMaster allocation because of a YARN queue or container-size constraint. A job with completed maps but stalled reducers may require examination of shuffle failures, reducer allocations, comparator semantics, or network capacity.

Start diagnosis with the effective configuration and command output, then use the service appropriate to the observed state:

1. Confirm the filesystem URI, input path, output path, permissions, and HDFS health with hdfs dfs, hdfs fsck, and NameNode reports.
2. Confirm DataNode and NodeManager registration, capacity, and health.
3. Inspect YARN application state, queue state, container diagnostics, and aggregated logs.
4. Inspect MapReduce counters, attempts, task diagnostics, and job history.
5. Change one reviewed configuration or operational condition, then verify the resulting state rather than treating a daemon restart as proof of recovery.

The progression preserves Hadoop’s boundaries: Common provides the shared runtime, HDFS provides storage, YARN provides resources, and MapReduce provides one batch execution model over both.