CSC352 MapReduce/Hadoop Class Notes

Outline

• History
• Based on Functional Programming
• Infrastructure, Streaming
• Submitting a Job
• Smith Cluster
• Example 1: Java
• Example 2: Python (streaming)
• Useful Commands

History

• Introduced in 2004
• MapReduce is a patented software framework introduced by Google to support distributed computing on large data sets on clusters of computers
• 2010 first conference: The First International Workshop on MapReduce and its Applications (MAPREDUCE'10). Interesting tidbit: nobody from Google on planning committee. Mostly INRIA

MapReduce Sibblings, and Related Projects

• Hadoop, open source. Written in Java, with a few routines in C. Wins Terabyte sorting competition in 2008

Apache Hadoop Wins Terabyte Sort Benchmark
One of Yahoo's Hadoop clusters sorted 1 terabyte of data in 209 seconds, which beat the previous record of 297 seconds in the annual general purpose (daytona) terabyte sort benchmark. The sort benchmark, which was created in 1998 by Jim Gray, specifies the input data (10 billion 100 byte records), which must be completely sorted and written to disk. This is the first time that either a Java or an open source program has won. Yahoo is both the largest user of Hadoop with 13,000+ nodes running hundreds of thousands of jobs a month and the largest contributor, although non-Yahoo usage and contributions are increasing rapidly.

Their implementation provided also to be highly scalable, and last year placed first place in the Terabyte Sort Benchmark. In order to do so, they used 910 nodes, each with 2 cores (a total of 1820 cores), and were able to keep the data entirely in memory across the nodes. With this many machines and their open-source MapReduce implementation, they were able to sort one terabyte of data in 209 seconds, in contrast to the previous record of 297 seconds. Not one to be left behind, Google showed that they were still quite ahead, claiming they could do the same with their closed-source implementation in only 68 seconds. The only details they note are that they use 1000 nodes, but since it’s not verifiable or mentioned on the official Sort Benchmark Home Page, it may not be worthwhile to discuss it further.

• Phoenix Another implementation of MapReduce, named Phoenix [2], has been created to facilitate optimal use of multi-core and multiprocessor systems. Written in C++, they sought to use the MapReduce programming paradigm to perform dense integer matrix multiplication as well as performing linear regression and principal components analysis on a matrix (amongst other applications). The following graph shows the speedup they achieve over varying number of cores:

• GPU/FPGA and MapReduce

MapReduce Based on Functional Programming Concepts

From

• Operations do not modify the data structures. New data created. The data are immutable.
• Original data unmodified
• Data flows are implicit in program design
• order of operations does not matter

Motivation

• Grand scale: data > TB, # servers > 1000.
• Automatic parallelization
• Fault Tolerant
• Simple and clean to program

Infrastructure

The Block Diagram

The Data is stored in the HDFS

• The data is held in a distributed file system, the Hadoop Distributed File System (HDFS)
• The goal of a DFS

A distributed file system is designed to hold a large amount of data and provide access to this data to many clients distributed across a network. There are a number of distributed file systems that solve this problem in different ways.

• HDFS is designed to store a very large amount of information (terabytes or petabytes). This requires spreading the data across a large number of machines. It also supports much larger file sizes than NFS.
• HDFS should store data reliably. If individual machines in the cluster malfunction, data should still be available.
• HDFS should provide fast, scalable access to this information.
• HDFS should integrate well with Hadoop MapReduce, allowing data to be read and computed upon locally when possible.

Figure 2.1: DataNodes holding blocks of multiple files with a replication factor of 2.
The NameNode maps the filenames onto the block ids.

The Computation

The images in this section are mostly taken from the excellent Yahoo tutorial (Module 4) on MapReduce

Programming Model

Material also taken from .

The WordCount Program (example)

The Map Function, simplified

mapper (filename, file-contents):
  for each word in file-contents:
    emit (word, 1)

The Reduce Function, simplified

reducer (word, values):
  sum = 0
  for each value in values:
    sum = sum + value
  emit (word, sum)

The Map and Reduce Java Blocks

  public static class MapClass extends MapReduceBase
    implements Mapper<LongWritable, Text, Text, IntWritable> {

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

    public void map(LongWritable key, Text value,
                    OutputCollector<Text, IntWritable> output,
                    Reporter reporter) throws IOException {
      String line = value.toString();
      StringTokenizer itr = new StringTokenizer(line);
      while (itr.hasMoreTokens()) {
        word.set(itr.nextToken());
        output.collect(word, one);
      }
    }
  }

  /**
   * A reducer class that just emits the sum of the input values.
   */
  public static class Reduce extends MapReduceBase
    implements Reducer<Text, IntWritable, Text, IntWritable> {

    public void reduce(Text key, Iterator<IntWritable> values,
                       OutputCollector<Text, IntWritable> output,
                       Reporter reporter) throws IOException {
      int sum = 0;
      while (values.hasNext()) {
        sum += values.next().get();
      }
      output.collect(key, new IntWritable(sum));
    }
  }

	Lab Experiment 1 Jump to the first Hadoop/MapReduce Lab #1!

Test: Running WordCound on 1, 2, 4, 8, and 16 identical files

• Setup: create 5 directories, one with 1 file, the other with 2 copies of the same file (with different names), one with 4 copies of the file, one with 8 copies, and 1 with 16 copies.

• the file contains "the quick red fox jumped over the lazy brown sleeping dog." which is 10 words long, and 60 bytes in size.

• The command to run hadoop on these directories is:

  hadoop jar /home/hadoop/352/dft/wordcount.jar org.myorg.WordCount dft_16fox dft_16fox-output

Output Log	1 file	2 files	4 files	8 files	16 files	comments
Counters: 17 File Systems HDFS bytes read HDFS bytes written Local bytes read Local bytes written Job Counters Launched reduce tasks Rack-local map tasks Launched map tasks Data-local map tasks Map-Reduce Framework Reduce input groups Combine output records Map input records Reduce output records Map output bytes Map input bytes Combine input records Map output records Reduce input records	17 401 75 121 644 1 6 12 6 10 10 1 10 103 60 11 11 10	17 430 75 236 874 1 6 12 6 10 20 2 10 206 120 22 22 20	17 476 75 466 1334 1 5 12 7 10 40 4 10 412 240 44 44 40	17 624 76 926 2390 1 2 16 14 10 80 8 10 824 480 88 88 80	17 960 85 1846 4230 1 1 16 15 10 160 16 10 1648 960 176 176 160	? ? ? ? ? ? ? ? ? 10 unique words 10 words times number of files number of files ? ? 60 bytes times number of files 10 unique words + 1 ? number of different words