Hadoop with Python

Map

The first phase of a MapReduce application is the map phase. Within the map phase, a function (called the mapper) processes a series of key-value pairs. The mapper sequentially processes each key-value pair individually, producing zero or more output key-value pairs (Figure 2-1).

As an example, consider a mapper whose purpose is to transform sentences into words. The input to this mapper would be strings that contain sentences, and the mapper’s function would be to split the sentences into words and output the words (Figure 2-2).

Shuffle and Sort

The second phase of MapReduce is the shuffle and sort. As the mappers begin completing, the intermediate outputs from the map phase are moved to the reducers. This process of moving output from the mappers to the reducers is known as shuffling.

Shuffling is handled by a partition function, known as the partitioner. The partitioner is used to control the flow of key-value pairs from mappers to reducers. The partitioner is given the mapper’s output key and the number of reducers, and returns the index of the intended reducer. The partitioner ensures that all of the values for the same key are sent to the same reducer. The default partitioner is hash-based. It computes a hash value of the mapper’s output key and assigns a partition based on this result.

The final stage before the reducers start processing data is the sorting process. The intermediate keys and values for each partition are sorted by the Hadoop framework before being presented to the reducer.

Reduce

The third phase of MapReduce is the reduce phase. Within the reducer phase, an iterator of values is provided to a function known as the reducer. The iterator of values is a nonunique set of values for each unique key from the output of the map phase. The reducer aggregates the values for each unique key and produces zero or more output key-value pairs (Figure 2-3).

As an example, consider a reducer whose purpose is to sum all of the values for a key. The input to this reducer is an iterator of all of the values for a key, and the reducer sums all of the values. The reducer then outputs a key-value pair that contains the input key and the sum of the input key values (Figure 2-4).

The next section describes a simple MapReduce application and its implementation in Python.

How It Works

The mapper and reducer are both executables that read input, line by line, from the standard input (stdin), and write output to the standard output (stdout). The Hadoop streaming utility creates a MapReduce job, submits the job to the cluster, and monitors its progress until it is complete.

When the mapper is initialized, each map task launches the specified executable as a separate process. The mapper reads the input file and presents each line to the executable via stdin. After the executable processes each line of input, the mapper collects the output from stdout and converts each line to a key-value pair. The key consists of the part of the line before the first tab character, and the value consists of the part of the line after the first tab character. If a line contains no tab character, the entire line is considered the key and the value is null.

When the reducer is initialized, each reduce task launches the specified executable as a separate process. The reducer converts the input key-value pair to lines that are presented to the executable via stdin. The reducer collects the executables result from stdout and converts each line to a key-value pair. Similar to the mapper, the executable specifies key-value pairs by separating the key and value by a tab character.

A Python Example

To demonstrate how the Hadoop streaming utility can run Python as a MapReduce application on a Hadoop cluster, the WordCount application can be implemented as two Python programs: mapper.py and reducer.py.

mapper.py is the Python program that implements the logic in the map phase of WordCount. It reads data from stdin, splits the lines into words, and outputs each word with its intermediate count to stdout. The code in Example 2-1 implements the logic in mapper.py.

#!/usr/bin/env python

import sys

# Read each line from stdin
for line in sys.stdin:

  # Get the words in each line
  words = line.split()

  # Generate the count for each word
  for word in words:

     # Write the key-value pair to stdout to be processed by 
     # the reducer.
     # The key is anything before the first tab character and the 
     #value is anything after the first tab character.
     print '{0}\t{1}'.format(word, 1)

reducer.py is the Python program that implements the logic in the reduce phase of WordCount. It reads the results of mapper.py from stdin, sums the occurrences of each word, and writes the result to stdout. The code in Example 2-2 implements the logic in reducer.py.

#!/usr/bin/env python

import sys

curr_word = None
curr_count = 0

# Process each key-value pair from the mapper
for line in sys.stdin:

  # Get the key and value from the current line
  word, count = line.split('\t')

  # Convert the count to an int
  count = int(count)

  # If the current word is the same as the previous word, 
  # increment its count, otherwise print the words count 
  # to stdout
  if word == curr_word:
     curr_count += count
  else:

     # Write word and its number of occurrences as a key-value 
     # pair to stdout
     if curr_word:
        print '{0}\t{1}'.format(curr_word, curr_count)

     curr_word = word
     curr_count = count

# Output the count for the last word
if curr_word == word:
  print '{0}\t{1}'.format(curr_word, curr_count)

Before attempting to execute the code, ensure that the mapper.py and reducer.py files have execution permission. The following command will enable this for both files:

$ chmod a+x mapper.py reducer.py

Also ensure that the first line of each file contains the proper path to Python. This line enables mapper.py and reducer.py to execute as standalone executables. The value #!/usr/bin/env python should work for most systems, but if it does not, replace /usr/bin/env python with the path to the Python executable on your system.

To test the Python programs locally before running them as a MapReduce job, they can be run from within the shell using the echo and sort commands. It is highly recommended to test all programs locally before running them across a Hadoop cluster.

$ echo 'jack be nimble jack be quick' | ./mapper.py 
| sort -t 1 | ./reducer.py
be      2
jack    2
nimble  1
quick   1

Once the mapper and reducer programs are executing successfully against tests, they can be run as a MapReduce application using the Hadoop streaming utility. The command to run the Python programs mapper.py and reducer.py on a Hadoop cluster is as follows:

$ $HADOOP_HOME/bin/hadoop jar 
  $HADOOP_HOME/mapred/contrib/streaming/hadoop-streaming*.jar \
-files mapper.py,reducer.py \
-mapper mapper.py \
-reducer reducer.py \
-input /user/hduser/input.txt -output /user/hduser/output

The options used with the Hadoop streaming utility are listed in Table 2-1.

Installation

The installation of mrjob is simple; it can be installed with pip by using the following command:

$ pip install mrjob

Or it can be installed from source (a git clone):

$ python setup.py install

WordCount in mrjob

Example 2-3 uses mrjob to implement the WordCount algorithm.

from mrjob.job import MRJob

class MRWordCount(MRJob):

   def mapper(self, _, line):
      for word in line.split():
         yield(word, 1)

   def reducer(self, word, counts):
      yield(word, sum(counts))

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

To run the mrjob locally, the only thing needed is a body of text. To run the job locally and count the frequency of words within a file named input.txt, use the following command:

$ python word_count.py input.txt

The output depends on the contents of the input file, but should look similar to Example 2-4.

"be"    2
"jack"    2
"nimble"    1
"quick"    1

What Is Happening

The MapReduce job is defined as the class, MRWordCount. Within the mrjob library, the class that inherits from MRJob contains the methods that define the steps of the MapReduce job. The steps within an mrjob application are mapper, combiner, and reducer. The class inheriting MRJob only needs to define one of these steps.

The mapper() method defines the mapper for the MapReduce job. It takes key and value as arguments and yields tuples of (output_key, output_value). In the WordCount example (Example 2-4), the mapper ignored the input key and split the input value to produce words and counts.

The combiner() method defines the combiner for the MapReduce job. The combiner is a process that runs after the mapper and before the reducer. It receives, as input, all of the data emitted by the mapper, and the output of the combiner is sent to the reducer. The combiner’s input is a key, which was yielded by the mapper, and a value, which is a generator that yields all values yielded by one mapper that corresponds to the key. The combiner yields tuples of (output_key, output_value) as output.

The reducer() method defines the reducer for the MapReduce job. It takes a key and an iterator of values as arguments and yields tuples of (output_key, output_value). In Example 2-4, the reducer sums the value for each key, which represents the frequency of words in the input.

The final component of a MapReduce job written with the mrjob library is the two lines at the end of the file:

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

These lines enable the execution of mrjob; without them, the application will not work.

Executing mrjob

Executing a MapReduce application with mrjob is similar to executing any other Python program. The command line must contain the name of the mrjob application and the input file:

$ python mr_job.py input.txt

By default, mrjob writes output to stdout.

Multiple files can be passed to mrjob as inputs by specifying the filenames on the command line:

$ python mr_job.py input1.txt input2.txt input3.txt

mrjob can also handle input via stdin:

$ python mr_job.py < input.txt

By default, mrjob runs locally, allowing code to be developed and debugged before being submitted to a Hadoop cluster.

To change how the job is run, specify the -r/--runner option. Table 2-2 contains a description of the valid choices for the runner options.

Using the runner option allows the mrjob program to be run on a Hadoop cluster, with input being specified from HDFS:

$ python mr_job.py -r hadoop hdfs://input/input.txt

mrjob also allows applications to be run on EMR directly from the command line:

$ python mr_job.py -r emr s3://input-bucket/input.txt

Top Salaries

Example 2-5 uses mrjob to compute employee top annual salaries and gross pay. The dataset used is the salary information from the city of Baltimore for 2014.

from mrjob.job import MRJob
from mrjob.step import MRStep
import csv

cols = 'Name,JobTitle,AgencyID,Agency,HireDate,AnnualSalary,GrossPay'.split(',')

class salarymax(MRJob):

    def mapper(self, _, line):
        # Convert each line into a dictionary
        row = dict(zip(cols, [ a.strip() for a in csv.reader([line]).next()]))

        # Yield the salary
        yield 'salary', (float(row['AnnualSalary'][1:]), line)
        
        # Yield the gross pay
        try:
            yield 'gross', (float(row['GrossPay'][1:]), line)
        except ValueError:
            self.increment_counter('warn', 'missing gross', 1)

    def reducer(self, key, values):
        topten = []

        # For 'salary' and 'gross' compute the top 10
        for p in values:
            topten.append(p)
            topten.sort()
            topten = topten[-10:]

        for p in topten:
            yield key, p

    combiner = reducer

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

Use the following command to execute the MapReduce job on Hadoop:

$ python top_salary.py -r hadoop hdfs:///user/hduser/input/salaries.csv

WordCount in Detail

This section describes each Pig Latin statement in the wordcount.pig script.

The first statement loads data from the filesystem and stores it in the relation records:

records = LOAD '/user/hduser/input/input.txt';

The second statement splits each line of text using the TOKENIZE function and eliminates nesting using the FLATTEN operator:

terms = FOREACH records GENERATE FLATTEN(TOKENIZE((chararray)$0)) AS word;

The third statement uses the GROUP operator to group the tuples that have the same field:

grouped_terms = GROUP terms BY word;

The fourth statement iterates over all of the terms in each bag and uses the COUNT function to return the sum:

word_counts = FOREACH grouped_terms GENERATE COUNT(terms), group;

The fifth and final statement stores the results in HDFS:

STORE word_counts INTO '/user/hduser/output/pig_wordcount'

Execution Modes

Pig has two execution modes: local and MapReduce.

Running Pig in local mode only requires a single machine. Pig will run on the local host and access the local filesystem. To run Pig in local mode, use the -x local flag:

$ pig -x local ...

Running Pig in MapReduce mode requires access to a Hadoop cluster. MapReduce mode executes Pig statements and jobs on the cluster and accesses HDFS. To run Pig in MapReduce mode, simply call Pig from the command line or use the -x mapreduce flag:

$ pig ...
or
$ pig -x mapreduce ...

Interactive Mode

Pig can be run interactively in the Grunt shell. To invoke the Grunt shell, simply call Pig from the command line and specify the desired execution mode. The following example starts the Grunt shell in local mode:

pig -x local
...
grunt>

Once the Grunt shell is initialized, Pig Latin statements can be entered and executed in an interactive manner. Running Pig interactively is a great way to learn Pig.

The following example reads /etc/passwd and displays the usernames from within the Grunt shell:

grunt> A = LOAD '/etc/passwd' using PigStorage(':');
grunt> B = FOREACH A GENERATE $0 as username;
grunt> DUMP B;

Batch Mode

Batch mode allows Pig to execute Pig scripts in local or MapReduce mode.

The Pig Latin statements in Example 3-3 read a file named passwd and use the STORE operator to store the results in a directory called user_id.out. Before executing this script, ensure that /etc/passwd is copied to the current working directory if Pig will be run in local mode, or to HDFS if Pig will be executed in MapReduce mode.

A = LOAD 'passwd' using PigStorage(':');
B = FOREACH A GENERATE $0 as username;
STORE B INTO 'user_id.out';

Use the following command to execute the user_id.pig script on the local machine:

$ pig -x local user_id.pig

Statements

Statements are the basic constructs used to process data in Pig. Each statement is an operator that takes a relation as an input, performs a transformation on that relation, and produces a relation as an output. Statements can span multiple lines, but all statements must end with a semicolon (;).

The general form of each Pig script is as follows:

1. A LOAD statement that reads the data from the filesystem
2. One or more statements to transform the data
3. A DUMP or STORE statement to view or store the results, respectively

Loading Data

The LOAD operator is used to load data from the system into Pig. The format of the LOAD operator is as follows:

LOAD 'data' [USING function] [AS schema];

Where 'data' is the name of the file or directory, in quotes, to be loaded. If a directory name is not specified, all of the files within the directory are loaded.

The USING keyword is optional and is used to specify a function to parse the incoming data. If the USING keyword is omitted, the default loading function, PigStorage, is used. The default delimiter is the tab character ('\t').

The AS keyword allows a schema to be defined for the data being loaded. Schemas enable names and datatypes to be declared for individual fields. The following example defines a schema for the data being loaded from the file input.txt. If no schema is defined, the fields are not named and default to type bytearray.

A = LOAD 'students' AS (name:chararray, age:int);

DUMP A;
(john,21,3.89)
(sally,19,2.56)
(alice,22,3.76)
(doug,19,1.98)
(susan,26,3.25)

Transforming Data

Pig contains many operators that enable complex transforming of data. The most common operators are FILTER, FOREACH, and GROUP.

A = LOAD 'students' AS (name:chararray, age:int, gpa:float);

DUMP A;
(john,21,3.89)
(sally,19,2.56)
(alice,22,3.76)
(doug,19,1.98)
(susan,26,3.25)

R = FILTER A BY age >= 20;

DUMP R;
(john,21,3.89)
(alice,22,3.76)
(susan,26,3.25)

R = FILTER A BY (age >= 20) AND (gpa > 3.5)

DUMP R;
(john,21,3.89)
(alice,22,3.76)

R = FOREACH A GENERATE *;

DUMP R;
(john,21,3.89)
(sally,19,2.56)
(alice,22,3.76)
(doug,19,1.98)
(susan,26,3.25)

R = FOREACH A GENERATE age, gpa;

DUMP R;
(21,3.89)
(19,2.56)
(22,3.76)
(19,1.98)
(26,3.25)

B = GROUP A BY age;

DUMP B;
(19,{(doug,19,1.98),(sally,19,2.56)})
(21,{(john,21,3.89)})
(22,{(alice,22,3.76)})
(26,{(susan,26,3.25)})

DESCRIBE B;
B: {group: int,A: {(name: chararray,age: int,gpa: float)}}

ILLUSTRATE B;
------------------------------------------------------------
| B  | group:int  | A:bag{:tuple(name:chararray,            |
                    age:int,gpa:float)}                     | 
-------------------------------------------------------------
|    | 19         | {(sally, 19, 2.56), (doug, 19, 1.98)}   | 
-------------------------------------------------------------

C = FOREACH B GENERATE group, A.name;

DUMP C;
(19,{(doug),(sally)})
(21,{(john)})
(22,{(alice)})
(26,{(susan)})

Storing Data

The STORE operator is used to execute previous Pig statements and store the results on the filesystem. The format of the STORE operator is as follows:

STORE alias INTO 'directory' [USING function];

Where alias is the name of the relation to store, and 'directory' is the name of the storage directory, in quotes. If the directory already exists, the STORE operation will fail. The output files will be named part-nnnnn and are written to the specified directory.

The USING keyword is optional and is used to specify a function to store the data. If the USING keyword is omitted, the default storage function, PigStorage, is used. The following example specifies the PigStorage function to store a file with pipe-delimited fields:

A = LOAD 'students' AS (name:chararray, age:int, gpa:float);

DUMP A;
(john,21,3.89)
(sally,19,2.56)
(alice,22,3.76)
(doug,19,1.98)
(susan,26,3.25)

STORE A INTO 'output' USING PigStorage('|');

CAT output;
john|21|3.89
sally|19|2.56
alice|22|3.76
doug|19|1.98
susan|26|3.25

The provided Pig Latin statements are great general-purpose computing constructs, but are not capable of expressing complex algorithms. The next section describes how to extend the functionality of Pig with Python.

Registering a UDF

Before a Python UDF can be used in a Pig script, it must be registered so Pig knows where to look when the UDF is called. To register a Python UDF file, use Pig’s REGISTER statement:

REGISTER 'udfs/myudf.py' USING streaming_python AS my_udf;

Once the UDF is registered, it can be called from within the Pig script:

relation = FOREACH data GENERATE my_udf.function(field);

In this example the UDF, referenced as my_udf, contains a function called function.

A Simple Python UDF

A simple Python UDF, located in pig/udfs/my_first_udf.py, that returns the integer value 1 each time it is called, is shown in Example 3-5.

from pig_util import outputSchema

@outputSchema('value:int')
def return_one():
   """
   Return the integer value 1
   """
   return 1

Some important things to note in this Python script are the from statement on the first line, and the output decorator, @outputSchema decorator, on the third line. These lines enable the Python UDF to define an alias and datatype for the data being returned from the UDF.

The Pig script in Example 3-6 registers the Python UDF and calls the return_one() function in a FOREACH statement.

REGISTER 'udfs/my_first_udf.py' USING streaming_python AS pyudfs;

A = LOAD '../resources/input.txt';
B = FOREACH A GENERATE pyudfs.return_one();
DUMP B;

When the Pig script is executed, it will generate an integer value 1 for each line in the input file. Use the following command to execute the script (sample output is shown as well):

$ pig -x local simple_udf.pig
...
(1)
(1)
(1)

String Manipulation

Python UDFs are an easy way of extending Pig’s functionality and an easy way to transform and process data.

The Python UDF in Example 3-7 contains two functions: reverse() and num_chars(). The reverse() function takes in a chararray and returns the chararray in reverse order. The num_chars() function takes in a chararray and returns the number of characters in the chararray.

from pig_util import outputSchema

@outputSchema('word:chararray')
def reverse(word):
   """
   Return the reverse text of the provided word
   """
   return word[::-1]

@outputSchema('length:int')
def num_chars(word):
   """
   Return the length of the provided word
   """
   return len(word)

The Pig script in Example 3-8 loads a text file and applies the reverse() and num_chars() Python functions to each unique word.

REGISTER 'udfs/string_funcs.py' USING streaming_python AS string_udf;

-- Load the data from the file system
records = LOAD '../resources/input.txt';

-- Split each line of text and eliminate nesting
terms = FOREACH records GENERATE FLATTEN(TOKENIZE((chararray) $0)) AS word;

-- Group similar terms
grouped_terms = GROUP terms BY word;

-- Count the number of tuples in each group
unique_terms = FOREACH grouped_terms GENERATE group as word;

-- Calculate the number of characters in each term
term_length = FOREACH unique_terms GENERATE word, string_udf.num_chars(word) as length;

-- Display the terms and their length
DUMP term_length;

-- Reverse each word  
reverse_terms = FOREACH unique_terms GENERATE word, string_udf.reverse(word) as reverse_word;

-- Display the terms and the reverse terms
DUMP reverse_terms;

Use the following command to execute the script (sample output shown):

$ pig -x local playing_with_words.pig
...
(be,2)
(the,3)
(jack,4)
(over,4)
(quick,5)
(jumped,6)
(nimble,6)
(candlestick,11)
...
(be,eb)
(the,eht)
(jack,kcaj)
(over,revo)
(quick,kciuq)
(jumped,depmuj)
(nimble,elbmin)
(candlestick,kcitseldnac)

Most Recent Movies

The following example uses movie data from the groupLens datasets and external libraries to calculate the 10 most recent movies.

The Python UDF in Example 3-9 contains two functions: parse_title() and days_since_release(). The parse_title() function uses Python’s regular expression module to remove the release year from a movie’s title. The days_since_release() function uses the datetime module to calculate the number of days between the current day and a movie’s release date.

from pig_util import outputSchema
from datetime import datetime
import re


@outputSchema('title:chararray')
def parse_title(title):
   """
   Return the title without the year
   """
   return re.sub(r'\s*\(\d{4}\)','', title)

@outputSchema('days_since_release:int')
def days_since_release(date):
   """
   Calculate the number of days since the titles release
   """
   if date is None:
      return None

   today = datetime.today()
   release_date = datetime.strptime(date, '%d-%b-%Y')
   delta = today - release_date
   return delta.days

The Pig script in Example 3-10 uses the Python UDFs to determine the 10 most recent movies.

REGISTER 'udfs/movies_udf.py' USING streaming_python AS movies_udf;

-- Load the data from the file system
records = LOAD '../resources/movies' USING PigStorage('|') 
   AS (id:int, title:chararray, release_date:chararray);

-- Parse the titles and determine how many days since the release date
titles = FOREACH records GENERATE movies_udf.parse_title(title), movies_udf.days_since_release(release_date);

-- Order the movies by the time since release
most_recent = ORDER titles BY days_since_release ASC;

-- Get the ten most recent movies
top_ten = LIMIT most_recent 10;

-- Display the top ten most recent movies
DUMP top_ten;

The following command is used to execute the script (sample output shown):

$ pig -x local recent_movies.pig 
...
(unknown,)
(Apt Pupil,6183)
(Mighty, The,6197)
(City of Angels,6386)
(Big One, The,6393)
(Lost in Space,6393)
(Mercury Rising,6393)
(Spanish Prisoner, The,6393)
(Hana-bi,6400)
(Object of My Affection, The,6400)

WordCount Described

This section describes the transformations being applied in the word_count.py Spark script.

The first statement creates a SparkContext object. This object tells Spark how and where to access a cluster:

sc = SparkContext(appName='SparkWordCount')

The second statement uses the SparkContext to load a file from HDFS and store it in the variable input_file:

input_file = sc.textFile('/user/hduser/input/input.txt')

The third statement performs multiple transformations on the input data. Spark automatically parallelizes these transformations to run across multiple machines:

counts = input_file.flatMap(lambda line: line.split()) \
                   .map(lambda word: (word, 1)) \
                   .reduceByKey(lambda a, b: a + b)

The fourth statement stores the results to HDFS:

counts.saveAsTextFile('/user/hduser/output')

The fifth statement shuts down the SparkContext:

sc.stop()

Interactive Shell

In the Spark shell, the SparkContext is created when the shell launches. The SparkContext is held in the variable sc. The master for the interactive shell can be set by using the --master argument when the shell is launched. To start an interactive shell, run the pyspark command:

$ pyspark --master local[4]
...
Welcome to
      ____              __
     / __/__  ___ _____/ /__
    _\ \/ _ \/ _ `/ __/  '_/
   /__ / .__/\_,_/_/ /_/\_\   version 1.5.0
      /_/
Using Python version 2.7.10 (default, Jul 13 2015 12:05:58)
SparkContext available as sc, HiveContext available as sqlContext.
>>>

For a complete list of options, run pyspark --help.

Self-Contained Applications

Self-contained applications must first create a SparkContext object before using any Spark methods. The master can be set when the SparkContext() method is called:

sc = SparkContext(master='local[4]')

To execute self-contained applications, they must be submitted to the spark-submit script. The spark-submit script contains many options; to see a complete listing, run spark-submit --help from the command line:

$ spark-submit --master local spark_app.py

Creating RDDs from Collections

RDDs can be created from a Python collection by calling the SparkContext.parallelize() method. The elements of the collection are copied to form a distributed dataset that can be operated on in parallel. The following example creates a parallelized collection from a Python list:

>>> data = [1, 2, 3, 4, 5]
>>> rdd = sc.parallelize(data)
>>> rdd.glom().collect()
...
[[1, 2, 3, 4, 5]]

The RDD.glom() method returns a list of all of the elements within each partition, and the RDD.collect() method brings all the elements to the driver node. The result, [[1, 2, 3, 4, 5]], is the original collection within a list.

To specify the number of partitions an RDD should be created with, a second argument can be passed to the parallelize() method. The following example creates an RDD from the same Python collection in the previous example, except this time four partitions are created:

>>> rdd = sc.parallelize(data, 4)
>>> rdd.glom().collect()
...
[[1], [2], [3], [4, 5]]

Using the glom() and collect() methods, the RDD created in this example contains four inner lists: [1], [2], [3], and [4, 5]. The number of inner lists represents the number of partitions within the RDD.

Creating RDDs from External Sources

RDDs can also be created from files using the SparkContext.textFile() method. Spark can read files residing on the local filesystem, any storage source supported by Hadoop, Amazon S3, and so on. Spark supports text files, SequenceFiles, any other Hadoop InputFormat, directories, compressed files, and wildcards, e.g., my/directory/*.txt. The following example creates a distributed dataset from a file located on the local filesystem:

>>> distFile = sc.textFile('data.txt')
>>> distFile.glom().collect()
...
[[u'jack be nimble', u'jack be quick', u'jack jumped over the candlestick']]

As before, the glom() and collect() methods allow the RDD to be displayed in its partitions. This result shows that distFile only has a single partition.

Similar to the parallelize() method, the textFile() method takes a second parameter that specifies the number of partitions to create. The following example creates an RDD with three partitions from the input file:

>>> distFile = sc.textFile('data.txt', 3)
>>> distFile.glom().collect()
...
[[u'jack be nimble', u'jack be quick'], [u'jack jumped over the candlestick'], []]

RDD Operations

RDDs support two types of operations: transformations and actions. Transformations create new datasets from existing ones, and actions run a computation on the dataset and return results to the driver program.

Transformations are lazy: that is, their results are not computed immediately. Instead, Spark remembers all of the transformations applied to a base dataset. Transformations are computed when an action requires a result to be returned to the driver program. This allows Spark to operate efficiently and only transfer the results of the transformations before an action.

By default, transformations may be recomputed each time an action is performed on it. This allows Spark to efficiently utilize memory, but it may utilize more processing resources if the same transformations are constantly being processed. To ensure a transformation is only computed once, the resulting RDD can be persisted in memory using the RDD.cache() method.

>>> lines = sc.textFile('data.txt')
>>> line_lengths = lines.map(lambda x: len(x))
>>> document_length = line_lengths.reduce(lambda x,y: x+y)
>>> print document_length
59

>>> line_lengths.persist()

lambda a, b: a + b

lambda x: len(x)

>>> data = [1, 2, 3, 4, 5, 6]
>>> rdd = sc.parallelize(data)
>>> map_result = rdd.map(lambda x: x * 2)
>>> map_result.collect()
[2, 4, 6, 8, 10, 12]

>>> data = [1, 2, 3, 4, 5, 6]
>>> filter_result = rdd.filter(lambda x: x % 2 == 0)
>>> filter_result.collect()
[2, 4, 6]

>>> data = [1, 2, 3, 2, 4, 1]
>>> rdd = sc.parallelize(data)
>>> distinct_result = rdd.distinct()
>>> distinct_result.collect()
[4, 1, 2, 3]

>>> data = [1, 2, 3, 4]
>>> rdd = sc.parallelize(data)
>>> map = rdd.map(lambda x: [x, pow(x,2)])
>>> map.collect()
[[1, 1], [2, 4], [3, 9], [4, 16]]

>>> rdd = sc.parallelize()
>>> flat_map = rdd.flatMap(lambda x: [x, pow(x,2)])
>>> flat_map.collect()
[1, 1, 2, 4, 3, 9, 4, 16]

>>> data = [1, 2, 3]
>>> rdd = sc.parallelize(data)
>>> rdd.reduce(lambda a, b: a * b)
6

>>> data = [1, 2, 3]
>>> rdd = sc.parallelize(data)
>>> rdd.take(2)
[1, 2]

>>> data = [1, 2, 3, 4, 5]
>>> rdd = sc.parallelize(data)
>>> rdd.collect()
[1, 2, 3, 4, 5]

>>> rdd.take(100).collect()

>>> data = [6,1,5,2,4,3]
>>> rdd = sc.parallelize(data)
>>> rdd.takeOrdered(4, lambda s: -s)
[6, 5, 4, 3]

Tasks

Tasks are the sequences of actions that comprise a Luigi workflow. Each task declares its dependencies on targets created by other tasks. This enables Luigi to create dependency chains that ensure a task will not be executed until all of the dependent tasks and all of the dependencies for those tasks are satisfied.

Figure 5-1 depicts a workflow highlighting Luigi tasks and their dependencies.

Target

Targets are the inputs and outputs of a task. The most common targets are files on a disk, files in HDFS, or records in a database. Luigi wraps the underlying filesystem operations to ensure that interactions with targets are atomic. This allows a workflow to be replayed from the point of failure without having to replay any of the already successfully completed tasks.

Parameters

Parameters allow the customization of tasks by enabling values to be passed into a task from the command line, programmatically, or from another task. For example, the name of a task’s output may be determined by a date passed into the task through a parameter.

Task.requires

The requires() method is used to specify a task’s dependencies. The WordCount task requires the output of the InputFile task:

 def requires(self):
    return InputFile(self.input_file)

It is important to note that the requires() method cannot return a Target object. In this example, the Target object is wrapped in the InputFile task. Calling the InputFile task with the self.input_file argument enables the input_file parameter to be passed to the InputFile task.

Task.output

The output() method returns one or more Target objects. The InputFile task returns the Target object that was the input for the WordCount task:

def output(self):
   return luigi.LocalTarget(self.input_file)

The WordCount task returns the Target object that was the output for the workflow:

def output(self):
   return luigi.LocalTarget(self.output_file)

Task.run

The run() method contains the code for a task. After the requires() method completes, the run() method is executed. The run() method for the WordCount task reads data from the input file, counts the number of occurrences, and writes the results to an output file:

def run(self):
   count = {}

   ifp = self.input().open('r')

   for line in ifp:
      for word in line.strip().split():
         count[word] = count.get(word, 0) + 1

   ofp = self.output().open('w')
   for k, v in count.items():
      ofp.write('{}\t{}\n'.format(k, v))
   ofp.close()

The input() and output() methods are helper methods that allow the task to read and write to Target objects in the requires() and output() methods, respectively.

Parameters

Parameters enable values to be passed into a task, customizing the task’s execution. The WordCount task contains two parameters: input_file and output_file:

class WordCount(luigi.Task):
   input_file = luigi.Parameter()
   output_file = luigi.Parameter(default='/tmp/wordcount')

Default values can be set for parameters by using the default argument.

Luigi creates a command-line parser for each Parameter object, enabling values to be passed into the Luigi script on the command line, e.g., --input-file input.txt and --output-file /tmp/output.txt.

Execution

To enable execution from the command line, the following lines must be present in the application:

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

This will enable Luigi to read commands from the command line.

The following command will execute the workflow, reading from input.txt and storing the results in /tmp/wordcount.txt:

$ python wordcount.py WordCount \
--local-scheduler \
--input-file input.txt \
--output-file /tmp/wordcount.txt

Configuration File

The examples in this section require a Luigi configuration file, client.cfg, to specify the location of the Hadoop streaming jar and the path to the Pig home directory. The config files should be in the current working directory, and an example of a config file is shown in Example 5-2.

[hadoop]
streaming-jar: /usr/lib/hadoop-xyz/hadoop-streaming-xyz-123.jar

[pig]
home: /usr/lib/pig

MapReduce in Luigi

Luigi scripts can control the execution of MapReduce jobs on a Hadoop cluster by using Hadoop streaming (Example 5-3).

import luigi
import luigi.contrib.hadoop
import luigi.contrib.hdfs

class InputFile(luigi.ExternalTask):
   """
   A task wrapping the HDFS target
   """
   input_file = luigi.Parameter()

   def output(self):
      """
      Return the target on HDFS
      """
      return luigi.contrib.hdfs.HdfsTarget(self.input_file)

class WordCount(luigi.contrib.hadoop.JobTask):
   """
   A task that uses Hadoop streaming to perform WordCount
   """
   input_file = luigi.Parameter()
   output_file = luigi.Parameter()

   # Set the number of reduce tasks
   n_reduce_tasks = 1

   def requires(self):
      """
      Read from the output of the InputFile task
      """
      return InputFile(self.input_file)

   def output(self):
      """
      Write the output to HDFS
      """
      return luigi.contrib.hdfs.HdfsTarget(self.output_file)

   def mapper(self, line):
      """
      Read each line and produce a word and 1
      """
      for word in line.strip().split():
         yield word, 1

   def reducer(self, key, values):
      """
      Read each word and produce the word and the sum of 
      its values
      """
      yield key, sum(values)

if __name__ == '__main__':
   luigi.run(main_task_cls=WordCount)

Luigi comes packaged with support for Hadoop streaming. The task implementing the MapReduce job must subclass luigi⁠.con⁠trib​.hadoop.JobTask. The mapper() and reducer() methods can be overridden to implement the map and reduce methods of a MapReduce job.

The following command will execute the workflow, reading from /user/hduser/input.txt and storing the results in /user/hduser/wordcount on HDFS:

$ python luigi_mapreduce.py --local-scheduler \
--input-file /user/hduser/input/input.txt \
--output-file /user/hduser/wordcount

Pig in Luigi

Luigi can be used to control the execution of Pig on a Hadoop cluster (Example 5-4).

import luigi
import luigi.contrib.pig
import luigi.contrib.hdfs

class InputFile(luigi.ExternalTask):
   """
   A task wrapping the HDFS target
   """
   input_file = luigi.Parameter()

   def output(self):
      return luigi.contrib.hdfs.HdfsTarget(self.input_file)

class WordCount(luigi.contrib.pig.PigJobTask):
   """
   A task that uses Pig to perform WordCount
   """
   input_file = luigi.Parameter()
   output_file = luigi.Parameter()
   script_path = luigi.Parameter(default='pig/wordcount.pig')

   def requires(self):
      """
      Read from the output of the InputFile task
      """
      return InputFile(self.input_file)

   def output(self):
      """
      Write the output to HDFS
      """
      return luigi.contrib.hdfs.HdfsTarget(self.output_file)     

   def pig_parameters(self):
      """
      A dictionary of parameters to pass to pig
      """
      return {'INPUT': self.input_file, 'OUTPUT': self.output_file}

   def pig_options(self):
      """
      A list of options to pass to pig
      """
      return ['-x', 'mapreduce']

   def pig_script_path(self):
      """
      The path to the pig script to run
      """
      return self.script_path

if __name__ == '__main__':
   luigi.run(main_task_cls=WordCount)

Luigi comes packaged with support for Pig. The task implementing the Pig job must subclass luigi.contrib.hadoop.PigJobTask. The pig_script_path() method is used to define the path to the Pig script to run. The pig_options() method is used to define the options to pass to the Pig script. The pig_parameters() method is used to pass parameters to the Pig script.

The following command will execute the workflow, reading from /user/hduser/input.txt and storing the results in /user/hduser/output on HDFS. The --script-path parameter is used to define the Pig script to execute:

$ python luigi_pig.py --local-scheduler \
--input-file /user/hduser/input/input.txt \
--output-file /user/hduser/output \
--script-path pig/wordcount.pig