Block 1 copy 1 → Node A (Rack 1)
Block 1 copy 2 → Node B (Rack 1) ← Different nodes, ideally different racks!
Block 1 copy 3 → Node C (Rack 2)

Traditional:
 [Data in Server A] ──────────────────→ [Compute Server B]
 (Huge data transfer over network!)

Hadoop (Data Locality):
 [Data in Node A] + [MapTask runs ON Node A]
 (No network transfer — computation happens locally!)

Replica 1 → Node on Rack 1 (local access)
Replica 2 → Different Node on Rack 1 (same rack, faster)
Replica 3 → Node on Rack 2 (different rack — survives rack failure!)

File: /logs/2024/jan/access.log
 → Block 1 (offset 0 to 128MB)
 → Replica A: DataNode 7 (192.168.1.7)
 → Replica B: DataNode 12 (192.168.1.12)
 → Replica C: DataNode 31 (192.168.1.31)
 → Block 2 (offset 128MB to 256MB)
 → Replica A: DataNode 3
 → Replica B: DataNode 19
 → Replica C: DataNode 25
 ... (8190 more blocks)

HADOOP 1:
 Cluster → MapReduce ONLY
 (Want Spark? Build another cluster )

HADOOP 2 (with YARN):
 Cluster → YARN (Resource Manager)
 ├── MapReduce jobs
 ├── Apache Spark jobs 
 ├── Apache Flink jobs
 ├── Apache Storm jobs
 └── Any other YARN-compatible framework

order_001, product_A, 299.00
order_002, product_B, 150.00
order_003, product_A, 299.00
order_004, product_C, 499.00
order_005, product_A, 299.00
... (10 million more rows)

MAP phase (runs on each block in parallel):
 Each mapper reads its portion of orders and emits:
 (product_A, 299.00)
 (product_B, 150.00)
 (product_A, 299.00) ← Same product from different orders
 ...

REDUCE phase (one reducer per product):
 Reducer for "product_A" receives: [299.00, 299.00, 299.00, ...]
 Sums them: total = ₹8,97,00,000 for the day
 Outputs: (product_A, 8970000.00)

SELECT product_id, SUM(amount) AS daily_total
FROM orders
WHERE date = '2024-01-01'
GROUP BY product_id
ORDER BY daily_total DESC;

TRADITIONAL SCALE-UP: HADOOP SCALE-OUT:
─────────────────────── ───────────────────────────────────
 Need more capacity? Need more capacity?
 Buy a BIGGER server Add MORE servers

 Cost: Exponential Cost: Linear
 (bigger = much more expensive) (add $5K servers, not $500K ones)

 Limit: Hardware ceiling Limit: Almost none!
 (can't add infinite RAM) (Google has millions of nodes)

 Failure: Catastrophic Failure: Graceful
 (1 server dies = everything gone) (1 of 10,000 dies = 0.01% impact)

1 node → process 1 TB in 10 hours
2 nodes → process 1 TB in 5 hours (2x faster, 2x cheaper per TB)
10 nodes → process 1 TB in 1 hour (10x faster)
100 nodes → process 1 TB in 6 min (100x faster)
1,000 nodes → process 1 TB in 36 sec (1,000x faster!)

The Hadoop Streaming Contract:
 Mapper program reads lines from stdin → emits "key\tvalue\n" to stdout
 Reducer program reads "key\tvalue" lines → emits "resultkey\tresultvalue\n" to stdout

Hadoop handles everything else:
 • Splitting input files
 • Sorting intermediate key/value pairs
 • Routing sorted pairs to the right reducer
 • Writing output to HDFS

#!/usr/bin/env python3
import sys

for line in sys.stdin: # Read from stdin
 line = line.strip()
 words = line.split()
 for word in words:
 print(f"{word}\t1") # Emit word → 1 to stdout

#!/usr/bin/env python3
import sys

current_word = None
current_count = 0

for line in sys.stdin: # Read sorted key-value pairs
 line = line.strip()
 word, count = line.split('\t', 1)
 count = int(count)

 if current_word == word:
 current_count += count
 else:
 if current_word:
 print(f"{current_word}\t{current_count}") # Emit final count
 current_word = word
 current_count = count

if current_word:
 print(f"{current_word}\t{current_count}")

hadoop jar $HADOOP_HOME/tools/lib/hadoop-streaming-*.jar \
 -input /data/books/*.txt \
 -output /output/wordcount \
 -mapper "python3 mapper.py" \
 -reducer "python3 reducer.py" \
 -file mapper.py \
 -file reducer.py

Problem: EditLog grows forever as you make file system changes
 FsImage snapshot gets staler and staler

Secondary NameNode (periodically):
 1. Downloads the current FsImage + EditLog from NameNode
 2. Merges them into a new up-to-date FsImage
 3. Uploads the fresh FsImage back to the NameNode
 4. NameNode now starts a fresh (small) EditLog

Result: EditLog stays small, FsImage stays fresh!

// ─── Setup ───────────────────────────────────────────────────────
Configuration conf = new Configuration();
// conf.set("fs.defaultFS", "hdfs://namenode:8020"); // for real cluster
FileSystem fs = FileSystem.get(conf);

// ─── Write a file ────────────────────────────────────────────────
Path outputPath = new Path("/user/nitish/hello.txt");
FSDataOutputStream out = fs.create(outputPath);
out.writeUTF("Hello HDFS! ");
out.close();
System.out.println("File written to HDFS successfully!");

// ─── Read a file ─────────────────────────────────────────────────
FSDataInputStream in = fs.open(outputPath);
String content = in.readUTF();
in.close();
System.out.println("Read from HDFS: " + content);

// ─── List directory contents ─────────────────────────────────────
FileStatus[] files = fs.listStatus(new Path("/user/nitish/"));
for (FileStatus file : files) {
 System.out.printf("%-10s %s%n",
 file.isDirectory() ? "[DIR]" : "[FILE]",
 file.getPath().getName()
 );
}

// ─── Check if path exists ────────────────────────────────────────
if (fs.exists(outputPath)) {
 System.out.println("File exists!");
}

// ─── Delete a file ───────────────────────────────────────────────
fs.delete(outputPath, false); // false = don't delete recursively

// Local filesystem
FileSystem local = FileSystem.get(URI.create("file:///"), conf);

// HDFS
FileSystem hdfs = FileSystem.get(URI.create("hdfs://namenode:8020/"), conf);

// Amazon S3
FileSystem s3 = FileSystem.get(URI.create("s3a://my-bucket/"), conf);

// All use the same FileSystem API! 

INPUT: Books stored in HDFS (War and Peace, Harry Potter, etc.)

QUESTION: "How many times does each word appear?"

→ MAP: Read lines → emit (word, 1) for each word
→ SHUFFLE: Group all (word, 1) pairs by word
→ REDUCE: Sum all 1s for each word → emit (word, total_count)

OUTPUT: (the, 8543), (and, 5212), (dragon, 127), ...

import java.io.IOException;
import java.util.StringTokenizer;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.*;
import org.apache.hadoop.mapreduce.*;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

public class WordCount {

 // ─── MAPPER ───────────────────────────────────────────────────
 // Input: (byte_offset, "To be or not to be")
 // Output: (word, 1) pairs for each word found
 public static class TokenizerMapper
 extends Mapper<Object, Text, Text, IntWritable> {

 private final static IntWritable one = new IntWritable(1); // Reused
 private Text word = new Text(); // Reused

 @Override
 public void map(Object key, Text value, Context context)
 throws IOException, InterruptedException {

 StringTokenizer itr = new StringTokenizer(value.toString());
 while (itr.hasMoreTokens()) {
 word.set(itr.nextToken().toLowerCase()); // lowercase for consistency
 context.write(word, one); // Emit (word, 1)
 }
 }
 }

 // ─── REDUCER ──────────────────────────────────────────────────
 // Input: ("the", [1, 1, 1, 1, 1, ...]) — all 1s for "the"
 // Output: ("the", 8543)
 public static class IntSumReducer
 extends Reducer<Text, IntWritable, Text, IntWritable> {

 private IntWritable result = new IntWritable();

 @Override
 public void reduce(Text key, Iterable<IntWritable> values, Context context)
 throws IOException, InterruptedException {

 int sum = 0;
 for (IntWritable val : values) {
 sum += val.get(); // Add up all the 1s
 }
 result.set(sum);
 context.write(key, result); // Emit (word, total_count)
 }
 }

 // ─── DRIVER ───────────────────────────────────────────────────
 public static void main(String[] args) throws Exception {
 Configuration conf = new Configuration();
 Job job = Job.getInstance(conf, "word count");

 job.setJarByClass(WordCount.class); // Which JAR to send to cluster
 job.setMapperClass(TokenizerMapper.class); // Our mapper
 job.setCombinerClass(IntSumReducer.class); // Local pre-aggregation (optimization!)
 job.setReducerClass(IntSumReducer.class); // Our reducer

 job.setOutputKeyClass(Text.class); // Output key type: String
 job.setOutputValueClass(IntWritable.class); // Output value type: Integer

 FileInputFormat.addInputPath(job, new Path(args[0])); // Input: /data/books/
 FileOutputFormat.setOutputPath(job, new Path(args[1])); // Output: /output/wordcount/

 System.exit(job.waitForCompletion(true) ? 0 : 1); // Wait and report success/failure
 }
}

WITHOUT Combiner:
 Mapper on Node A processes 1,000 lines of text
 → Emits 5,000 (word, 1) pairs
 → Sends ALL 5,000 pairs over network to reducers

 The word "the" appears 400 times in this mapper's chunk
 → Sends 400 × ("the", 1) over network = 400 messages!

WITH Combiner:
 Mapper on Node A processes 1,000 lines of text
 → Emits 5,000 (word, 1) pairs
 → Combiner runs locally: ("the", [1,1,1,...400 times...]) → ("the", 400)
 → Sends 1 message instead of 400!

Network savings: ~400x reduction for this word!

# Compile
javac -classpath $(hadoop classpath) WordCount.java
jar cf wordcount.jar WordCount*.class

# Run on cluster
hadoop jar wordcount.jar WordCount /data/books/ /output/wordcount/

# View results
hdfs dfs -cat /output/wordcount/part-r-00000 | sort -k2 -nr | head -20

the 8543
and 5212
to 4891
of 4212
a 3876
...
dragon 127

File: /data/books/warandpeace.txt (512 MB)
→ Split into 4 InputSplits of 128 MB each
→ 4 MapTasks will be created (one per split)
→ Each MapTask is assigned to a node where that block lives! (data locality)

MapTask 1 reads its 128 MB chunk line by line:
 Line: "To be or not to be that is the question"
 Emits: (to,1) (be,1) (or,1) (not,1) (to,1) (be,1) (that,1) (is,1) (the,1) (question,1)

 Line: "The dragon flew over the mountains"
 Emits: (the,1) (dragon,1) (flew,1) (over,1) (the,1) (mountains,1)

 ... continues for all lines in its 128MB chunk

Default HashPartitioner:
 "the" → hash("the") = 12345 → 12345 % 3 (reducers) = 0 → goes to Reducer 0
 "dragon" → hash("dragon") = 67890 → 67890 % 3 = 1 → goes to Reducer 1
 "and" → hash("and") = 55555 → 55555 % 3 = 2 → goes to Reducer 2

Reducer 0 says: "Give me all partition-0 data from ALL mappers"
→ Fetches from MapTask 1 (on Node A): [("the", 400 tuples)]
→ Fetches from MapTask 2 (on Node B): [("the", 350 tuples)]
→ Fetches from MapTask 3 (on Node C): [("the", 280 tuples)]
→ Fetches from MapTask 4 (on Node D): [("the", 313 tuples)]

All data for Reducer 0, now on one node:
 Unsorted: (the,400) (and,200) (that,150) (the,350) (and,250) ...
 Sort by key: (and,200) (and,250) (that,150) (the,400) (the,350) ...
 Group: (and, [200, 250]) (that, [150]) (the, [400, 350]) ...

reduce("the", [400, 350, 280, 313]) → sum = 1343 → output: ("the", 1343)
reduce("and", [200, 250, ...]) → sum = ... → output: ("and", total)

Scenario: MapTask on Node 7 throws OutOfMemoryError

Hadoop response:
 1. NodeManager reports task failure to ApplicationMaster
 2. AM marks the attempt as failed
 3. AM requests a new container for a fresh attempt (elsewhere)
 4. New MapTask re-reads the SAME InputSplit from HDFS (using another replica)
 5. Job continues!

Default: Up to 4 attempts before the job fails entirely

Scenario: NodeManager on Node 15 stops responding

Hadoop response:
 1. ApplicationMaster notices missing heartbeats (after timeout)
 2. AM marks Node 15 as lost
 3. All tasks running on Node 15 are re-assigned to other nodes
 4. If Node 15 had map outputs needed for reduce phase:
 → Those map tasks are RE-RUN on other nodes!

HDFS response (separate):
 1. NameNode notices missing DataNode heartbeats
 2. Identifies all blocks that lost a replica
 3. Schedules re-replication on surviving nodes
 4. Replication factor restored automatically!

Scenario: Job has 100 map tasks. 99 finish in 5 minutes.
 MapTask 73 on Node 42 is only 60% done after 10 minutes.

 Why? Node 42 might have: slower disk, network issues,
 background processes stealing CPU, or just bad luck.

Without speculative execution:
 Entire job waits for slow MapTask 73! ⏰ Wastes everyone's time.

With speculative execution:
 1. Framework detects MapTask 73 is "significantly behind average"
 2. Launches a DUPLICATE attempt on a different, free node
 3. Whichever finishes first → its output is used 
 4. The loser is killed

Result: Job completes nearly as fast as the fastest 99 tasks!

Job Queue: [Big Job A (needs 80% cluster) → Small Job B → Small Job C]

Timeline:
 t=0: Job A starts. Uses 80% of cluster.
 t=5h: Job A finishes. 
 t=5h: Job B starts (waited 5 hours! )
 t=5h+: Job C starts after B

Problem: One huge job blocks everyone else!

Queue: [Long Job A, Short Job B, Short Job C] submitted simultaneously

Fair Scheduler behavior:
 t=0: All 3 jobs get 33% of cluster resources each
 t=10m: Job B finishes early! Resources freed.
 t=10m: Jobs A and C now each get 50% 
 t=20m: Job C finishes! Resources freed.
 t=20m: Job A gets 100% — finishes faster!

Result: No job starves. Short jobs complete quickly!

Configuration:
 Engineering team: 60% of cluster
 Marketing team: 30% of cluster
 Shared pool: 10% of cluster

Behavior:
 If Marketing has no jobs running → their 30% can be borrowed by others
 When Marketing submits a job → they get their 30% back (preemption)
 Engineering is ALWAYS guaranteed their 60%, no matter what!

NodeManager on each machine reports:
 "I have 64 GB RAM and 16 CPU cores available"

Resource requests:
 MapTask: 1 core, 2 GB RAM (typical)
 ReduceTask: 1 core, 4 GB RAM (typical, needs more for merging)

So this NodeManager can run:
 Up to 16 map tasks simultaneously (CPU limited)
 Or up to 32 map tasks (if 2GB each, RAM limited at 64GB)

[──────────────────────────────────────────────────]
 ↑ spill point (80%) ↑ fill point

 Map output is written to this circular buffer.
 When 80% full → spill thread sorts + writes to disk in background.
 Map task continues writing to the remaining 20%.

Reducer 0 needs partition 0 data from ALL 1,000 map tasks.
It opens HTTP connections to all 1,000 NodeManagers simultaneously
and pulls its partition data. (This is where network is saturated!)

AS data arrives from mappers, it's merged with previously arrived data.
Final result: One big sorted file with all keys for this reducer.
(the, [1,1,1,...8543 times...])
(dragon, [1,1,...127 times...])

User's reduce() method is finally called.
It receives one (key, Iterable<values>) at a time.
Output written to HDFS!

1. Container started by NodeManager (its own JVM process!)
 ↓
2. Task receives its InputSplit info (or reducer partition info)
 ↓
3. Map task: reads records from InputFormat (line by line for TextInputFormat)
 OR Reduce task: pulls shuffle data from mappers
 ↓
4. User's map()/reduce() code executes
 ↓
5. Output written (to buffer/disk for maps, to HDFS for reduces)
 ↓
6. Heartbeats sent to ApplicationMaster every few seconds:
 "I'm 45% done. Still alive! "
 ↓
7. Task completes → container released back to YARN pool

Map-Reduce Framework
 Map input records=10000000 ← How many records were processed
 Map output records=45823912 ← How many (k,v) pairs emitted
 Map output bytes=366591296 ← Size of map output
 Map output materialized bytes= ← After compression
 Input split bytes=128000000 ← Size of this task's input
 Combine input records= ← Records fed to combiner
 Combine output records= ← Records output from combiner
 Reduce input records= ← Records fed to reducer
 Reduce shuffle bytes= ← Bytes transferred in shuffle
 ...

ResourceManager Web UI: http://<rm-host>:8088
 → Click on your Application ID
 → Click on a finished/failed task
 → Click "logs"

Or via command line:
yarn logs -applicationId application_12345_0001

// TextOutputFormat (default): writes "key\tvalue\n"
job.setOutputFormatClass(TextOutputFormat.class);

// SequenceFileOutputFormat: writes binary (for chaining jobs)
job.setOutputFormatClass(SequenceFileOutputFormat.class);

// Multiple output files based on condition
MultipleOutputs<Text, IntWritable> mos = new MultipleOutputs<>(context);
mos.write("highValue", key, value, "high-value-products");
mos.write("lowValue", key, value, "low-value-products");

public class SalesRecord implements Writable {
 private Text productId = new Text();
 private DoubleWritable amount = new DoubleWritable();

 @Override
 public void write(DataOutput out) throws IOException {
 productId.write(out);
 amount.write(out);
 }

 @Override
 public void readFields(DataInput in) throws IOException {
 productId.readFields(in);
 amount.readFields(in);
 }
}

Rule: Combiner = Mini-Reducer that runs LOCALLY on each map node
Works for: SUM, MIN, MAX, COUNT (associative + commutative ops)
Does NOT work for: AVERAGE, MEDIAN (order/grouping dependent)

Example where combiner fails:
 Average temperature across all cities:
 Combiner computes local averages → reducer averages those averages
 = WRONG (unequal sample sizes)
 Instead: combiner sums totals + counts separately → reducer divides

// TotalOrderPartitioner: ensures output is globally sorted
// Example: sort all sales records by amount in total order
job.setPartitionerClass(TotalOrderPartitioner.class);

// CustomPartitioner: route specific keys to specific reducers
public class RegionPartitioner extends Partitioner<Text, IntWritable> {
 @Override
 public int getPartition(Text key, IntWritable value, int numPartitions) {
 String region = key.toString().split("_")[0];
 if (region.equals("NORTH")) return 0;
 if (region.equals("SOUTH")) return 1;
 return 2; // EAST, WEST, etc.
 }
}

// Define an enum for your counters
public enum MyCounters {
 RECORDS_PROCESSED,
 NULL_VALUES_FOUND,
 SUSPICIOUS_TRANSACTIONS,
 HIGH_VALUE_ORDERS
}

// In your Mapper:
public void map(Object key, Text value, Context context) {
 context.getCounter(MyCounters.RECORDS_PROCESSED).increment(1);

 String amount = value.toString().split(",")[2];
 if (amount == null || amount.isEmpty()) {
 context.getCounter(MyCounters.NULL_VALUES_FOUND).increment(1);
 return; // skip this record
 }

 if (Double.parseDouble(amount) > 100000) {
 context.getCounter(MyCounters.HIGH_VALUE_ORDERS).increment(1);
 }

 // ... continue processing
}

Counters: 5
 MyCounters
 HIGH_VALUE_ORDERS=1247
 NULL_VALUES_FOUND=32
 RECORDS_PROCESSED=10000000
 SUSPICIOUS_TRANSACTIONS=89

// In your Driver:
job.addCacheFile(new URI("hdfs:///reference/customer_names.txt"));

// In your Mapper's setup() method (runs once per task, not once per record):
@Override
protected void setup(Context context) throws IOException {
 URI[] cacheFiles = context.getCacheFiles();
 BufferedReader reader = new BufferedReader(
 new FileReader(new File(cacheFiles[0].getPath()))
 );
 String line;
 while ((line = reader.readLine()) != null) {
 String[] parts = line.split(",");
 customerNames.put(parts[0], parts[1]); // Load into HashMap
 }
}

// In your map() method:
public void map(Object key, Text value, Context context) {
 String customerId = extractCustomerId(value);
 String name = customerNames.get(customerId); // Fast in-memory lookup!
 // ...
}

<!-- core-site.xml: Basic cluster settings -->
<configuration>
 <property>
 <name>fs.defaultFS</name>
 <value>hdfs://namenode:8020</value>
 <!-- This tells clients where the NameNode is -->
 </property>
 <property>
 <name>hadoop.tmp.dir</name>
 <value>/data/hadoop/tmp</value>
 </property>
</configuration>

<!-- hdfs-site.xml: HDFS-specific settings -->
<configuration>
 <property>
 <name>dfs.replication</name>
 <value>3</value> <!-- Default replication factor -->
 </property>
 <property>
 <name>dfs.blocksize</name>
 <value>134217728</value> <!-- 128 MB in bytes -->
 </property>
 <property>
 <name>dfs.namenode.name.dir</name>
 <value>/data/hdfs/namenode</value> <!-- Where NameNode stores metadata -->
 </property>
 <property>
 <name>dfs.datanode.data.dir</name>
 <value>/data/hdfs/datanode</value> <!-- Where DataNodes store blocks -->
 </property>
</configuration>

<!-- mapred-site.xml: MapReduce settings -->
<configuration>
 <property>
 <name>mapreduce.framework.name</name>
 <value>yarn</value> <!-- Use YARN for resource management -->
 </property>
 <property>
 <name>mapreduce.map.memory.mb</name>
 <value>2048</value> <!-- 2 GB per map task -->
 </property>
 <property>
 <name>mapreduce.reduce.memory.mb</name>
 <value>4096</value> <!-- 4 GB per reduce task -->
 </property>
</configuration>

<!-- yarn-site.xml: YARN resource manager settings -->
<configuration>
 <property>
 <name>yarn.resourcemanager.hostname</name>
 <value>resourcemanager</value>
 </property>
 <property>
 <name>yarn.nodemanager.resource.memory-mb</name>
 <value>65536</value> <!-- 64 GB RAM per NodeManager -->
 </property>
 <property>
 <name>yarn.nodemanager.resource.cpu-vcores</name>
 <value>16</value> <!-- 16 cores per NodeManager -->
 </property>
</configuration>

# Format the NameNode (ONLY do this once when setting up fresh!)
hdfs namenode -format

# Start HDFS (NameNode + DataNode)
start-dfs.sh

# Start YARN (ResourceManager + NodeManager)
start-yarn.sh

# Verify everything is running
jps
# Should show:
# 12345 NameNode
# 12789 DataNode
# 13456 SecondaryNameNode
# 14123 ResourceManager
# 14567 NodeManager

HDFS Status: http://localhost:9870 (Browse files, check health)
YARN Jobs: http://localhost:8088 (Running jobs, queue status)
Job History: http://localhost:19888 (Completed job details and logs)

Without security: Any user can read/write any HDFS path! 
With Kerberos:
 - Every user and service must authenticate with a Kerberos ticket
 - HDFS enforces Unix-style permissions (owner, group, others)
 - All data transfers can be encrypted (transparent)

Additional security layers:
 - HDFS encryption zones: Data encrypted on disk 
 - SSL/TLS: Data encrypted in transit 
 - Ranger: Fine-grained access control (table-level, column-level!)
 - Atlas: Data lineage and governance

# File Operations
hdfs dfs -ls /user/nitish/ # List directory
hdfs dfs -put localfile.txt /hdfs/path/ # Upload to HDFS
hdfs dfs -get /hdfs/path/ ./local/ # Download from HDFS
hdfs dfs -cat /hdfs/file.txt # Print file contents
hdfs dfs -mkdir /new/directory # Create directory
hdfs dfs -rm /hdfs/file.txt # Delete file
hdfs dfs -rm -r /hdfs/directory/ # Delete directory recursively
hdfs dfs -mv /old/path /new/path # Move file
hdfs dfs -cp /source /destination # Copy file

# Information
hdfs dfs -du -h /user/ # Disk usage (human readable)
hdfs dfs -df -h # HDFS free/used space
hdfs fsck / # Filesystem health check
hdfs dfs -stat %r /path # Replication factor of a file

# YARN Commands
yarn application -list # All running applications
yarn application -kill application_ID # Kill a job
yarn logs -applicationId app_ID # View job logs