Hadoop: The Definitive Guide

HDFS is a filesystem designed for storing very large files with streaming data access patterns(write-once, read-many-times pattern), running on clusters of commodity hardware.

HDFS blocks(>64M) are large compared to disk blocks, and the reason is to minimize the cost of seeks. Map tasks in MapReduce normally operate on one block at a time, so if you have too few tasks (fewer than nodes in the cluster), your jobs will run slower than they could otherwise.

An HDFS cluster has two types of node operating in a master-worker pattern: a namenode (the master) and a number of datanodes (workers). The namenode manages the filesystem namespace. It maintains the filesystem tree and the metadata for all the files and directories in the tree. Datanodes are the workhorses of the filesystem. They store and retrieve blocks when they are told to (by clients or the namenode), and they report back to the namenode periodically with lists of blocks that they are storing.

After a successful return from sync() , HDFS guarantees that the data written up to that point in the file is persisted and visible to all new readers.

Hadoop Archives , or HAR files, are a file archiving facility that packs files into HDFS blocks more efficiently, thereby reducing namenode memory usage while still allowing transparent access to files.

Read data flow of HDFS

One important aspect of this design is that the client contacts datanodes directly to retrieve data and is guided by the namenode to the best datanode for each block. This design allows HDFS to scale to a large number of concurrent clients, since the data traffic is spread across all the datanodes in the cluster. Hadoop takes a simple approach in which the network is represented as a tree and the distance between two nodes is the sum of their distances to their closest common ancestor.

Write data flow of HDFS

As the client writes data (step 3), DFSOutputStream splits it into packets, which it writes to an internal queue, called the data queue . The data queue is consumed by the Data Streamer , whose responsibility it is to ask the namenode to allocate new blocks by picking a list of suitable datanodes to store the replicas. The list of datanodes forms a pipeline—we’ll assume the replication level is three, so there are three nodes in the pipeline. The DataStreamer streams the packets to the first datanode in the pipeline, which stores the packet and forwards it to the second datanode in the pipeline. DFSOutputStream also maintains an internal queue of packets that are waiting to be acknowledged by datanodes, called the ack queue . A packet is removed from the ack queue only when it has been acknowledged by all the datanodes in the pipeline (step 5). If a datanode fails while data is being written to it, then the following actions are taken, which are transparent to the client writing the data. First the pipeline is closed, and any packets in the ack queue are added to the front of the data queue so that datanodes that are downstream from the failed node will not miss any packets. The current block on the good datanodes is given a new identity, which is communicated to the namenode, so that the partial block on the failed datanode will be deleted if the failed datanode recovers later on. The failed datanode is removed from the pipeline and the remainder of the block’s data is written to the two good datanodes in the pipeline. The namenode notices that the block is under-replicated, and it arranges for a further replica to be created on another node. Subsequent blocks are then treated as normal.

Hadoop’s default strategy is to place the first replica on the same node as the client (for clients running outside the cluster, a node is chosen at random, although the system tries not to pick nodes that are too full or too busy). The second replica is placed on a different rack from the first ( off-rack ), chosen at random. The third replica is placed on the same rack as the second, but on a different node chosen at random. Further replicas are placed on random nodes on the cluster, although the system tries to avoid placing too many replicas on the same rack.

HDFS Federation, introduced in the 0.23 release series, allows a cluster to scale by adding namenodes, each of which manages a portion of the filesystem namespace. For example, one namenode might manage all the files rooted under /user , say, and a second namenode might handle files under /share .

The 0.23 release series of Hadoop remedies this situation by adding support for HDFS high-availability (HA). In this implementation there is a pair of namenodes in an activestandby configuration.

MapReduce is a programming model for data processing. MapReduce works by breaking the processing into two phases: the map phase and the reduce phase. Each phase has key-value pairs as input and output, the types of which may be chosen by the programmer. The programmer also specifies two functions: the map function and the reduce function.

Hadoop divides the input to a MapReduce job into fixed-size pieces called input splits, or just splits. Hadoop creates one map task for each split, which runs the userdefined map function for each record in the split. Hadoop does its best to run the map task on a node where the input data resides in HDFS. This is called the data locality optimization.

When there are multiple reducers, the map tasks partition their output, each creating one partition for each reduce task. There can be many keys (and their associated values) in each partition, but the records for every key are all in a single partition. The partitioning can be controlled by a user-defined partitioning function, but normally the default partitioner—which buckets keys using a hash function—works very well.

Map Reduce里的shuffle过程

package org.apache.hadoop.book.ch02.min;

import java.io.IOException;

import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

public class NewMinTemperature {
      static class NewMinTemperatureMapper
            extends Mapper<LongWritable, Text, Text, IntWritable> {
            private static final int MISSING = 9999;

            @Override
            public void map(LongWritable key, Text value,
                        Context context) throws IOException, InterruptedException {
                  String line = value.toString();
                  String year = line.substring(15, 19);
                  int airTemperature;
                  if(line.charAt(87) == '+'){
                        airTemperature = Integer.parseInt(line.substring(88, 92));
                  } else {
                        airTemperature = Integer.parseInt(line.substring(87, 92));
                  }
                  String quality = line.substring(92, 93);
                  if(airTemperature != MISSING && quality.matches("[01459]")){
                        context.write(new Text(year), new IntWritable(airTemperature));
                  }

            }
      }

      static class NewMinTemperatureReducer
            extends Reducer<Text, IntWritable, Text, IntWritable>{
            @Override
            public void reduce(Text key, Iterable<IntWritable> values,
                        Context context)
                        throws IOException, InterruptedException{
                  int minValue = Integer.MAX_VALUE;
                  for(IntWritable value : values){
                        minValue = Math.min(minValue, value.get());
                  }
                  context.write(key, new IntWritable(minValue));
            }
      }

      public static void main(String args[]) throws Exception{
            if(args.length != 2){
                  System.err.println("Usage: Min Temperature <input path> <output path>");
                  System.exit(-1);
            }
            Job job = new Job();
            job.setJarByClass(NewMinTemperature.class);

            FileInputFormat.addInputPath(job, new Path(args[0]));
            FileOutputFormat.setOutputPath(job, new Path(args[1]));

            job.setMapperClass(NewMinTemperatureMapper.class);
            job.setCombinerClass(NewMinTemperatureReducer.class);  //optional for combiner

            job.setReducerClass(NewMinTemperatureReducer.class);

            job.setOutputKeyClass(Text.class);
            job.setOutputValueClass(IntWritable.class);

            System.exit(job.waitForCompletion(true)? 0: 1);
      }
}