Processing Distributed Data

Using MapReduce

Data Science 310

Boston University

MapReduce

A framework for computation on large data sets that are fragmented and replicated across a cluster of machines.

  • spreads the computation across the machines, letting them work in parallel
  • tries to minimize the amount of data that is transferred between machines

The original version was Google’s MapReduce system.

  • An open-source version is part of the Hadoop project (used in HW).

Sample Problem: Totalling Customer Orders

  • Acme Widgets is a company that sells only one type of product.
  • Data set: a large collection of records about customer orders
    • fragmented and replicated across a cluster of machines
  • sample record:

  • Desired computation: For each customer, compute the total amount in that customer’s active orders.
  • Inefficient approach: Ship all of the data to one machine and compute the totals there.

Sample Problem: Totalling Customer Orders (cont.)

  • MapReduce does better using “divide-and-conquer” approach.
    • splits the collection of records into subcollections that are processed in parallel
  • For each subcollection, a mapper task maps the records to smaller key-value pairs – in this case, (cust_id, amount active).

  • These smaller pairs are distributed by cust_id to other tasks that again work in parallel.
  • These reducer tasks combine the pairs for a given cust_id to compute the per-customer totals:

Benefits of MapReduce

  • Parallel processing reduces overall computation time.
  • Less data is sent between machines.
  • the mappers often operate on local data
  • the key-value pairs sent to the reducers are smaller than the original records
  • an initial reduction can sometimes be done locally
    • example: compute local subtotals for each customer, then send those subtotals to the reducers
  • It provides fault tolerance.
    • if a given task fails or is too slow, re-execute it
  • The framework handles all of the hard/messy parts.
  • The user can just focus on the problem being solved!

MapReduce In General: Mapping

  • The system divides up the collection of input records, and assigns each subcollection Si to a mapper task Mj.

  • The mappers apply a map function to each record:
map(k, v):   # treat record as a key-value pair
    emit 0 or more new key-value pairs (k', v')
  • the resulting keys and values (the intermediate results) can have different types than the original ones
  • the input and intermediate keys do not have to be unique

MapReduce In General: Reducing

  • The system partitions the intermediate results by key, and assigns each range of keys to a reducer task Rk.

  • Key-value pairs with the same key are grouped together:
(k', v'0),  (k', v'1), (k', v'2)   →   (k', [v'0, v'1, v'2, ...])
  • so that all values for a given key are processed together
  • The reducers apply a reduce function to each (key, value-list):
reduce(k', [v'0, v'1, v'2, ...]):
    emit 0 or more key-value pairs (k", v")
  • the types of the (k”, v”) can be different from the (k’, v’)

MapReduce In General: Combining (Optional)

  • In some cases, the intermediate results can be aggregated locally using combiner tasks Cn.
  • Often, the combiners use the same reduce function as the reducers.
    • produces partial results that can then be combined
  • This cuts down on the data transferred to the reducers.

Hadoop MapReduce Framework

  • Implemented in Java
  • It also includes other, non-Java options for writing MapReduce applications.
  • We will write applications in python using the MRJob library

MRJob Framework

from mrjob.job import MRJob

class Example1(MRJob):

    def mapper(self, input_key, input_value):
        // your code goes here
        yield (output_key, output_value)

    def reducer(self, input_key, input_value):
        // your code goes here
        yield (output_key, output_value)

if __name__ == '__main__':
    Example1.run()

Example 1: Birth-Month Counter

  • The data: text file(s) containing person records that look like this
    • id,name,dob,email
    • where dob is in the form yyyy-mm-dd
  • The problem: Find the number of people born in each month.

Example 1: Birth-Month Counter (cont.)

  • map should:
    • extract the month from the person’s dob
    • emit a single key-value pair of the form (month string, 1)

  • The intermediate results are distributed by key to the reducers.
  • reduce should:
    • add up the 1s for a given month
    • emit a single key-value pair of the form (month string, total)

Mapper for Example 1

  • For data obtained from text files, the Mapper’s inputs will be key-value pairs in which:
    • value = a single line from one of the files
    • key = None
  • Hence we use _ to “throw away” the key
  • The map method will output pairs in which
    • key = a month string
    • value = 1
  • We use yield to return the value (like “return”)

Splitting a String

  • The str class includes a method named split().
    • breaks a string into component strings
  • takes a parameter indicating what delimiter should be used when performing the split
  • returns a str list containing the components
  • Example:
sentence = "How now brown cow?"
words = sentence.split(" ")
print(words[0])
print(words[3])
print(len(words))

would output:

How
cow?
4

Processing an Input Record in map

def mapper(self, _, line):
  • Recall: line represents one record.
    • for Example 1, it looks like: 111,Alan Turing,1912-06-23,al@aol.com
  • To extract the month string:
    • split line on the commas to get the fields:
fields = line.split(",")
  • similarly, split the date field on the hyphens to get its components
  • could we just split line on the hyphens?
  • no, because a person’s name could have hyphens in it

Processing the List of Values in reduce

  • counts is a list of “1”s
  • sum() adds up a list
  • Use yield rather than return
python lecture-ex-1.py lecture-data-1.txt
"12"    3
"06"    2
"03"    1

Example 2: Month with the Most Birthdays

  • The data: same as Example 1. Records of the form
    • id,name,dob,email
    • where dob is in the form yyyy-mm-dd
  • The problem: Find the month with the most birthdays.

Example 2: Month with the Most Birthdays (cont.)

  • map should behave as before:

  • reduce needs to:

  • add up the 1s for a given month
  • determine which month has the largest total
  • but… there can be multiple reducer tasks, each of which handles one subset of the months
  • each reducer can only determine the largest month in its subset
  • the solution: a chain of two MapReduce jobs

Example 2: Chaining Jobs

  • First job = count birth months as we did in Example 1
    • map1: person record → (birth month, 1)
    • reduce1: (birth month, [1, 1, …]) → (birth month, total)
  • The second job processes the results of the first job!
  • map2: (birth month, total) → (c, (birth month, total))
    • output key c = an arbitrary constant, used for all k-v pairs (we will use “None”)
  • output value = a pairing of a birth month and its total
  • ("06", 2) → (None, (2, 06))
  • ("12", 3) → (None, (3, 12))
  • ("03", 1) → (None, (1, 03))
  • because there is only one output key, there is only one reducer task!
  • reduce2: find the month with the most birthdays
    • (None, [(2, 06), (3, 12), (1, 03)]) → (3, 12)

Example 2: Chaining Jobs (cont.)

class Lecture_Ex_2(MRJob):
    def steps(self):
        return [
            MRStep(mapper = self.mapper_get_months,
                   reducer = self.reducer_count_months),
            MRStep(mapper = self.mapper_single_key,
                   reducer = self.reducer_max)]

    def mapper_get_months(self, _, line):
        words = line.split(',')
        year, month, day = words[2].split('-')
        yield (month, 1)

    def reducer_count_months(self, month, counts):
        yield (month, sum(counts))

    def mapper_single_key(self, month, counts):
        yield (None, (counts, month))

    def reducer_max(self, _, month_count_pairs):
        yield max(month_count_pairs)

if __name__ == '__main__':
    Lecture_Ex_2.run()