MapReduce is a programming model developed by Google for processing and generating large data sets on an array of commodity servers. Greenplum MapReduce allows programmers who are familiar with the MapReduce model to write map and reduce functions and submit them to the Greenplum Database parallel engine for processing.

You configure a Greenplum MapReduce job via a YAML-formatted configuration file, then pass the file to the Greenplum MapReduce program, gpmapreduce, for execution by the Greenplum Database parallel engine. The Greenplum Database system distributes the input data, runs the program across a set of machines, handles machine failures, and manages the required inter-machine communication.

Refer to gpmapreduce for details about running the Greenplum MapReduce program.

Parent topic: Querying Data

About the Greenplum MapReduce Configuration File

This section explains some basics of the Greenplum MapReduce configuration file format to help you get started creating your own Greenplum MapReduce configuration files. Greenplum uses the YAML 1.1 document format and then implements its own schema for defining the various steps of a MapReduce job.

All Greenplum MapReduce configuration files must first declare the version of the YAML specification they are using. After that, three dashes (---) denote the start of a document, and three dots (...) indicate the end of a document without starting a new one. (A document in this context is equivalent to a MapReduce job.) Comment lines are prefixed with a pound symbol (#). You can declare multiple Greenplum MapReduce documents/jobs in the same file:

  1. %YAML 1.1
  2. ---
  3. # Begin Document 1
  4. # ...
  5. ---
  6. # Begin Document 2
  7. # ...

Within a Greenplum MapReduce document, there are three basic types of data structures or nodes: scalars, sequences and mappings.

A scalar is a basic string of text indented by a space. If you have a scalar input that spans multiple lines, a preceding pipe ( | ) denotes a literal style, where all line breaks are significant. Alternatively, a preceding angle bracket ( > ) folds a single line break to a space for subsequent lines that have the same indentation level. If a string contains characters that have reserved meaning, the string must be quoted or the special character must be escaped with a backslash ( \ ).

  1. # Read each new line literally
  2. somekey: |   this value contains two lines
  3.    and each line is read literally
  4. # Treat each new line as a space
  5. anotherkey: >
  6.    this value contains two lines
  7.    but is treated as one continuous line
  8. # This quoted string contains a special character
  9. ThirdKey: "This is a string: not a mapping"

A sequence is a list with each entry in the list on its own line denoted by a dash and a space (-). Alternatively, you can specify an inline sequence as a comma-separated list within square brackets. A sequence provides a set of data and gives it an order. When you load a list into the Greenplum MapReduce program, the order is kept.

  1. # list sequence
  2. - this
  3. - is
  4. - a list
  5. - with
  6. - five scalar values
  7. # inline sequence
  8. [this, is, a list, with, five scalar values]

A mapping is used to pair up data values with identifiers called keys. Mappings use a colon and space (:) for each key: value pair, or can also be specified inline as a comma-separated list within curly braces. The key is used as an index for retrieving data from a mapping.

  1. # a mapping of items
  2. title: War and Peace
  3. author: Leo Tolstoy
  4. date: 1865
  5. # same mapping written inline
  6. {title: War and Peace, author: Leo Tolstoy, date: 1865}

Keys are used to associate meta information with each node and specify the expected node type (scalar, sequence or mapping).

The Greenplum MapReduce program processes the nodes of a document in order and uses indentation (spaces) to determine the document hierarchy and the relationships of the nodes to one another. The use of white space is significant. White space should not be used simply for formatting purposes, and tabs should not be used at all.

Refer to gpmapreduce.yaml for detailed information about the Greenplum MapReduce configuration file format and the keys and values supported.

Example Greenplum MapReduce Job

In this example, you create a MapReduce job that processes text documents and reports on the number of occurrences of certain keywords in each document. The documents and keywords are stored in separate Greenplum Database tables that you create as part of the exercise.

This example MapReduce job utilizes the untrusted plpythonu language; as such, you must run the job as a user with Greenplum Database administrative privileges.

  1. Log in to the Greenplum Database master host as the gpadmin administrative user and set up your environment. For example:

    1. $ ssh gpadmin@<gpmaster>
    2. gpadmin@gpmaster$ . /usr/local/greenplum-db/greenplum_path.sh

  2. Create a new database for the MapReduce example: For example:

    1. gpadmin@gpmaster$ createdb mapredex_db

  3. Start the psql subsystem, connecting to the new database:

    1. gpadmin@gpmaster$ psql -d mapredex_db

  4. Register the PL/Python language in the database. For example:

    1. mapredex_db=> CREATE EXTENSION plpythonu;

  5. Create the documents table and add some data to the table. For example:

    1. CREATE TABLE documents (doc_id int, url text, data text);
    2. INSERT INTO documents VALUES (1, 'http:/url/1', 'this is one document in the corpus');
    3. INSERT INTO documents VALUES (2, 'http:/url/2', 'i am the second document in the corpus');
    4. INSERT INTO documents VALUES (3, 'http:/url/3', 'being third never really bothered me until now');
    5. INSERT INTO documents VALUES (4, 'http:/url/4', 'the document before me is the third document');

  6. Create the keywords table and add some data to the table. For example:

    1. CREATE TABLE keywords (keyword_id int, keyword text);
    2. INSERT INTO keywords VALUES (1, 'the');
    3. INSERT INTO keywords VALUES (2, 'document');
    4. INSERT INTO keywords VALUES (3, 'me');
    5. INSERT INTO keywords VALUES (4, 'being');
    6. INSERT INTO keywords VALUES (5, 'now');
    7. INSERT INTO keywords VALUES (6, 'corpus');
    8. INSERT INTO keywords VALUES (7, 'is');
    9. INSERT INTO keywords VALUES (8, 'third');

  7. Construct the MapReduce YAML configuration file. For example, open a file named mymrjob.yaml in the editor of your choice and copy/paste the following large text block:

    1. # This example MapReduce job processes documents and looks for keywords in them.
    2. # It takes two database tables as input:
    3. # - documents (doc_id integer, url text, data text)
    4. # - keywords (keyword_id integer, keyword text)#
    5. # The documents data is searched for occurrences of keywords and returns results of
    6. # url, data and keyword (a keyword can be multiple words, such as "high performance # computing")
    7. %YAML 1.1
    8. ---
    9. VERSION: 1.0.0.2
    11. # Connect to Greenplum Database using this database and role
    12. DATABASE: mapredex_db
    13. USER: gpadmin
    15. # Begin definition section
    16. DEFINE:
    18.  # Declare the input, which selects all columns and rows from the
    19.  # 'documents' and 'keywords' tables.
    20.   - INPUT:
    21.       NAME: doc
    22.       TABLE: documents
    23.   - INPUT:
    24.       NAME: kw
    25.       TABLE: keywords
    26.   # Define the map functions to extract terms from documents and keyword
    27.   # This example simply splits on white space, but it would be possible
    28.   # to make use of a python library like nltk (the natural language toolkit)
    29.   # to perform more complex tokenization and word stemming.
    30.   - MAP:
    31.       NAME: doc_map
    32.       LANGUAGE: python
    33.       FUNCTION: |
    34.         i = 0            # the index of a word within the document
    35.         terms = {}# a hash of terms and their indexes within the document
    37.         # Lower-case and split the text string on space
    38.         for term in data.lower().split():
    39.           i = i + 1# increment i (the index)
    41.         # Check for the term in the terms list:
    42.         # if stem word already exists, append the i value to the array entry 
    43.         # corresponding to the term. This counts multiple occurrences of the word.
    44.         # If stem word does not exist, add it to the dictionary with position i.
    45.         # For example:
    46.         # data: "a computer is a machine that manipulates data" 
    47.         # "a" [1, 4]
    48.         # "computer" [2]
    49.         # "machine" [3]
    50.         # …
    51.           if term in terms:
    52.             terms[term] += ','+str(i)
    53.           else:
    54.             terms[term] = str(i)
    56.         # Return multiple lines for each document. Each line consists of 
    57.         # the doc_id, a term and the positions in the data where the term appeared.
    58.         # For example: 
    59.         #   (doc_id => 100, term => "a", [1,4]
    60.         #   (doc_id => 100, term => "computer", [2]
    61.         #    …
    62.         for term in terms:
    63.           yield([doc_id, term, terms[term]])
    64.       OPTIMIZE: STRICT IMMUTABLE
    65.       PARAMETERS:
    66.         - doc_id integer
    67.         - data text
    68.       RETURNS:
    69.         - doc_id integer
    70.         - term text
    71.         - positions text
    73.   # The map function for keywords is almost identical to the one for documents 
    74.   # but it also counts of the number of terms in the keyword.
    75.   - MAP:
    76.       NAME: kw_map
    77.       LANGUAGE: python
    78.       FUNCTION: |
    79.         i = 0
    80.         terms = {}
    81.         for term in keyword.lower().split():
    82.           i = i + 1
    83.           if term in terms:
    84.             terms[term] += ','+str(i)
    85.           else:
    86.             terms[term] = str(i)
    88.         # output 4 values including i (the total count for term in terms):
    89.         yield([keyword_id, i, term, terms[term]])
    90.       OPTIMIZE: STRICT IMMUTABLE
    91.       PARAMETERS:
    92.         - keyword_id integer
    93.         - keyword text
    94.       RETURNS:
    95.         - keyword_id integer
    96.         - nterms integer
    97.         - term text
    98.         - positions text
    100.   # A TASK is an object that defines an entire INPUT/MAP/REDUCE stage
    101.   # within a Greenplum MapReduce pipeline. It is like EXECUTION, but it is
    102.   # run only when called as input to other processing stages.
    103.   # Identify a task called 'doc_prep' which takes in the 'doc' INPUT defined earlier
    104.   # and runs the 'doc_map' MAP function which returns doc_id, term, [term_position]
    105.   - TASK:
    106.       NAME: doc_prep
    107.       SOURCE: doc
    108.       MAP: doc_map
    110.   # Identify a task called 'kw_prep' which takes in the 'kw' INPUT defined earlier
    111.   # and runs the kw_map MAP function which returns kw_id, term, [term_position]
    112.   - TASK:
    113.       NAME: kw_prep
    114.       SOURCE: kw
    115.       MAP: kw_map
    117.   # One advantage of Greenplum MapReduce is that MapReduce tasks can be
    118.   # used as input to SQL operations and SQL can be used to process a MapReduce task.
    119.   # This INPUT defines a SQL query that joins the output of the 'doc_prep' 
    120.   # TASK to that of the 'kw_prep' TASK. Matching terms are output to the 'candidate' 
    121.   # list (any keyword that shares at least one term with the document).
    122.   - INPUT:
    123.       NAME: term_join
    124.       QUERY: |
    125.         SELECT doc.doc_id, kw.keyword_id, kw.term, kw.nterms,
    126.                doc.positions as doc_positions,
    127.                kw.positions as kw_positions
    128.           FROM doc_prep doc INNER JOIN kw_prep kw ON (doc.term = kw.term)
    130.   # In Greenplum MapReduce, a REDUCE function is comprised of one or more functions.
    131.   # A REDUCE has an initial 'state' variable defined for each grouping key. that is 
    132.   # A TRANSITION function adjusts the state for every value in a key grouping.
    133.   # If present, an optional CONSOLIDATE function combines multiple 
    134.   # 'state' variables. This allows the TRANSITION function to be run locally at
    135.   # the segment-level and only redistribute the accumulated 'state' over
    136.   # the network. If present, an optional FINALIZE function can be used to perform 
    137.   # final computation on a state and emit one or more rows of output from the state.
    138.   #
    139.   # This REDUCE function is called 'term_reducer' with a TRANSITION function 
    140.   # called 'term_transition' and a FINALIZE function called 'term_finalizer'
    141.   - REDUCE:
    142.       NAME: term_reducer
    143.       TRANSITION: term_transition
    144.       FINALIZE: term_finalizer
    146.   - TRANSITION:
    147.       NAME: term_transition
    148.       LANGUAGE: python
    149.       PARAMETERS:
    150.         - state text
    151.         - term text
    152.         - nterms integer
    153.         - doc_positions text
    154.         - kw_positions text
    155.       FUNCTION: |
    157.         # 'state' has an initial value of '' and is a colon delimited set 
    158.         # of keyword positions. keyword positions are comma delimited sets of 
    159.         # integers. For example, '1,3,2:4:' 
    160.         # If there is an existing state, split it into the set of keyword positions
    161.         # otherwise construct a set of 'nterms' keyword positions - all empty
    162.         if state:
    163.           kw_split = state.split(':')
    164.         else:
    165.           kw_split = []
    166.           for i in range(0,nterms):
    167.             kw_split.append('')
    169.         # 'kw_positions' is a comma delimited field of integers indicating what
    170.         # position a single term occurs within a given keyword. 
    171.         # Splitting based on ',' converts the string into a python list.
    172.         # add doc_positions for the current term
    173.         for kw_p in kw_positions.split(','):
    174.           kw_split[int(kw_p)-1] = doc_positions
    176.         # This section takes each element in the 'kw_split' array and strings 
    177.         # them together placing a ':' in between each element from the array.
    178.         # For example: for the keyword "computer software computer hardware", 
    179.         # the 'kw_split' array matched up to the document data of 
    180.         # "in the business of computer software software engineers" 
    181.         # would look like: ['5', '6,7', '5', ''] 
    182.         # and the outstate would look like: 5:6,7:5:
    183.         outstate = kw_split[0]
    184.         for s in kw_split[1:]:
    185.           outstate = outstate + ':' + s
    186.         return outstate
    188.   - FINALIZE:
    189.       NAME: term_finalizer
    190.       LANGUAGE: python
    191.       RETURNS:
    192.         - count integer
    193.       MODE: MULTI
    194.       FUNCTION: |
    195.         if not state:
    196.           yield 0
    197.         kw_split = state.split(':')
    199.         # This function does the following:
    200.         # 1) Splits 'kw_split' on ':'
    201.         #    for example, 1,5,7:2,8 creates '1,5,7' and '2,8'
    202.         # 2) For each group of positions in 'kw_split', splits the set on ',' 
    203.         #    to create ['1','5','7'] from Set 0: 1,5,7 and 
    204.         #    eventually ['2', '8'] from Set 1: 2,8
    205.         # 3) Checks for empty strings
    206.         # 4) Adjusts the split sets by subtracting the position of the set 
    207.         #      in the 'kw_split' array
    208.         # ['1','5','7'] - 0 from each element = ['1','5','7']
    209.         # ['2', '8'] - 1 from each element = ['1', '7']
    210.         # 5) Resulting arrays after subtracting the offset in step 4 are
    211.         #    intersected and their overlapping values kept: 
    212.         #    ['1','5','7'].intersect['1', '7'] = [1,7]
    213.         # 6) Determines the length of the intersection, which is the number of 
    214.         # times that an entire keyword (with all its pieces) matches in the 
    215.         #    document data.
    216.         previous = None
    217.         for i in range(0,len(kw_split)):
    218.           isplit = kw_split[i].split(',')
    219.           if any(map(lambda(x): x == '', isplit)):
    220.             yield 0
    221.           adjusted = set(map(lambda(x): int(x)-i, isplit))
    222.           if (previous):
    223.             previous = adjusted.intersection(previous)
    224.           else:
    225.             previous = adjusted
    227.         # return the final count
    228.         if previous:
    229.           yield len(previous)
    231.    # Define the 'term_match' task which is then run as part 
    232.    # of the 'final_output' query. It takes the INPUT 'term_join' defined
    233.    # earlier and uses the REDUCE function 'term_reducer' defined earlier
    234.   - TASK:
    235.       NAME: term_match
    236.       SOURCE: term_join
    237.       REDUCE: term_reducer
    238.   - INPUT:
    239.       NAME: final_output
    240.       QUERY: |
    241.         SELECT doc.*, kw.*, tm.count
    242.         FROM documents doc, keywords kw, term_match tm
    243.         WHERE doc.doc_id = tm.doc_id
    244.           AND kw.keyword_id = tm.keyword_id
    245.           AND tm.count > 0
    247. # Execute this MapReduce job and send output to STDOUT
    248. EXECUTE:
    249.   - RUN:
    250.       SOURCE: final_output
    251.       TARGET: STDOUT

  8. Save the file and exit the editor.

  9. Run the MapReduce job. For example:

    1. gpadmin@gpmaster$ gpmapreduce -f mymrjob.yaml

    The job displays the number of occurrences of each keyword in each document to stdout.

Flow Diagram for MapReduce Example

The following diagram shows the job flow of the MapReduce job defined in the example:

MapReduce job flow