WIXLIB

ted files to be copied to the map reduce cluster-libjars comma separated list of jars specify comma separated jar files to include in the classpath.-archives comma separated list of archives specify commaseparated archives to be unarchived on the compute machines.The general command line syntax isbin/hadoop command [genericOptions] [commandOptions]The next section introduces a Python library that allows HDFS to beaccessed from within Python applications.SnakebiteSnakebite is a Python package, created by Spotify, that provides aPython client library, allowing HDFS to be accessed programmati‐cally from Python applications. The client library uses protobufmessages to communicate directly with the NameNode. The Snake‐bite package also includes a command-line interface for HDFS thatis based on the client library.This section describes how to install and configure the Snakebitepackage. Snakebite’s client library is explained in detail with multipleexamples, and Snakebite’s built-in CLI is introduced as a Pythonalternative to the hdfs dfs command.InstallationSnakebite requires Python 2 and python-protobuf 2.4.1 or higher.Python 3 is currently not supported.Snakebite is distributed through PyPI and can be installed usingpip: pip install snakebiteClient LibraryThe client library is written in Python, uses protobuf messages, andimplements the Hadoop RPC protocol for talking to the NameNode.This enables Python applications to communicate directly withHDFS and not have to make a system call to hdfs dfs.List Directory ContentsExample 1-1 uses the Snakebite client library to list the contents ofthe root directory in HDFS.Snakebite 7

Example 1-1. python/HDFS/list directory.pyfrom snakebite.client import Clientclient Client('localhost', 9000)for x in client.ls(['/']):print xThe most important line of this program, and every program thatuses the client library, is the line that creates a client connection tothe HDFS NameNode:client Client('localhost', 9000)The Client() method accepts the following parameters:host (string)Hostname or IP address of the NameNodeport (int)RPC port of the NameNodehadoop version (int)The Hadoop protocol version to be used (default: 9)use trash (boolean)Use trash when removing fileseffective use (string)Effective user for the HDFS operations (default: None or cur‐rent user)The host and port parameters are required and their values aredependent upon the HDFS configuration. The values for theseparameters can be found in the hadoop/conf/core-site.xml configura‐tion file under the property fs.defaultFS: property name fs.defaultFS /name value hdfs://localhost:9000 /value /property For the examples in this section, the values used for host and portare localhost and 9000, respectively.After the client connection is created, the HDFS filesystem can beaccessed. The remainder of the previous application used the lscommand to list the contents of the root directory in HDFS:8 Chapter 1: Hadoop Distributed File System (HDFS)

for x in client.ls(['/']):print xIt is important to note that many of methods in Snakebite returngenerators. Therefore they must be consumed to execute. The lsmethod takes a list of paths and returns a list of maps that containthe file information.Executing the list directory.py application yields the followingresults: python list directory.py{'group': u'supergroup', 'permission': 448,'access time':0L,'block replication':tion time': 1442752574936L, 'length': 0L,'owner': u'hduser', 'path': '/tmp'}{'group': u'supergroup', 'permission': 493,'access time':0L,'block replication':tion time': 1442742056276L, 'length': 0L,'owner': u'hduser', 'path': '/user'}'file type': 'd',0,'modifica'blocksize': 0L,'file type': 'd',0,'modifica'blocksize': 0L,Create a DirectoryUse the mkdir() method to create directories on HDFS.Example 1-2 creates the directories /foo/bar and /input on HDFS.Example 1-2. python/HDFS/mkdir.pyfrom snakebite.client import Clientclient Client('localhost', 9000)for p in client.mkdir(['/foo/bar', '/input'], create parent True):print pExecuting the mkdir.py application produces the following results: python mkdir.py{'path': '/foo/bar', 'result': True}{'path': '/input', 'result': True}The mkdir() method takes a list of paths and creates the specifiedpaths in HDFS. This example used the create parent parameter toensure that parent directories were created if they did not alreadyexist. Setting create parent to True is analogous to the mkdir -pUnix command.Snakebite 9

Deleting Files and DirectoriesDeleting files and directories from HDFS can be accomplished withthe delete() method. Example 1-3 recursively deletes the /fooand /bar directories, created in the previous example.Example 1-3. python/HDFS/delete.pyfrom snakebite.client import Clientclient Client('localhost', 9000)for p in client.delete(['/foo', '/input'], recurse True):print pExecuting the delete.py application produces the following results: python delete.py{'path': '/foo', 'result': True}{'path': '/input', 'result': True}Performing a recursive delete will delete any subdirectories and filesthat a directory contains. If a specified path cannot be found, thedelete method throws a FileNotFoundException. If recurse is notspecified and a subdirectory or file exists, DirectoryException isthrown.The recurse parameter is equivalent to rm -rf and should be usedwith care.Retrieving Data from HDFSLike the hdfs dfs command, the client library contains multiplemethods that allow data to be retrieved from HDFS. To copy filesfrom HDFS to the local filesystem, use the copyToLocal() method.Example 1-4 copies the file /input/input.txt from HDFS and places itunder the /tmp directory on the local filesystem.Example 1-4. python/HDFS/copy to local.pyfrom snakebite.client import Clientclient Client('localhost', 9000)for f in client.copyToLocal(['/input/input.txt'], '/tmp'):print fExecuting the copy to local.py application produces the followingresult:10 Chapter 1: Hadoop Distributed File System (HDFS)

python copy to local.py{'path': '/tmp/input.txt', 'source path': '/input/input.txt','result': True, 'error': ''}To simply read the contents of a file that resides on HDFS, thetext() method can be used. Example 1-5 displays the contentof /input/input.txt.Example 1-5. python/HDFS/text.pyfrom snakebite.client import Clientclient Client('localhost', 9000)for l in client.text(['/input/input.txt']):print lExecuting the text.py application produces the following results: python text.pyjack be nimblejack be quickjack jumped over the candlestickThe text() method will automatically uncompress and display gzipand bzip2 files.CLI ClientThe CLI client included with Snakebite is a Python command-lineHDFS client based on the client library. To execute the SnakebiteCLI, the hostname or IP address of the NameNode and RPC port ofthe NameNode must be specified. While there are many ways tospecify these values, the easiest is to create a .snakebiterc configura‐tion file. Example 1-6 contains a sample config with the NameNodehostname of localhost and RPC port of 9000.Example 1-6. /.snakebiterc{"config version": 2,"skiptrash": true,"namenodes": [{"host": "localhost", "port": 9000, "version": 9},]}Snakebite 11

The values for host and port can be found in the hadoop/conf/coresite.xml configuration file under the property fs.defaultFS.For more information on configuring the CLI, see the Snakebite CLIdocumentation online.UsageTo use the Snakebite CLI client from the command line, simply usethe command snakebite. Use the ls option to display the contentsof a directory: snakebite ls /Found 2 itemsdrwx------ hadoopdrwxr-xr-x- hadoopsupergroupsupergroup0 2015-09-20 14:36 /tmp0 2015-09-20 11:40 /userLike the hdfs dfs command, the CLI client supports many familiarfile manipulation commands (e.g., ls, mkdir, df, du, etc.).The major difference between snakebite and hdfs dfs is thatsnakebite is a pure Python client and does not need to load anyJava libraries to communicate with HDFS. This results in quickerinteractions with HDFS from the command line.CLI Command ReferenceThe following is a full listing of file manipulation commands possi‐ble with the snakebite CLI client. This listing can be displayed fromthe command line by specifying snakebite without any arguments.To view help with a specific command, use snakebite [cmd] -help, where cmd is a valid snakebite command.snakebite [general options] cmd [arguments]general options:-D --debugShow debug information-V --versionHadoop protocol version (default:9)-h --helpshow help-j --jsonJSON output-n --namenodenamenode host-p --portnamenode RPC port (default: 8020)-v --verDisplay snakebite versioncommands:cat [paths]chgrp grp [paths]chmod mode [paths]chown owner:grp [paths]copyToLocal [paths] dst12copy source paths to stdoutchange groupchange file mode (octal)change ownercopy paths to local Chapter 1: Hadoop Distributed File System (HDFS)

count [paths]dfdu [paths]get file dstgetmerge dir dstls [paths]mkdir [paths]mkdirp [paths]mv [paths] dstrm [paths]rmdir [dirs]serverdefaultssetrep rep [paths]stat [paths]tail pathtest pathtext path [paths]touchz [paths]usage cmd file system destinationdisplay stats for pathsdisplay fs statsdisplay disk usage statisticscopy files to localfile system destinationconcatenates files in source dirinto destination local filelist a pathcreate directoriescreate directories and theirparentsmove paths to destinationremove pathsdelete a directoryshow server informationset replication factorstat informationdisplay last kilobyte of thefile to stdouttest a pathoutput file in text formatcreates a file of zero lengthshow cmd usageto see command-specific options use: snakebite [cmd] --helpChapter SummaryThis chapter introduced and described the core concepts of HDFS.It explained how to interact with the filesystem using the built-inhdfs dfs command. It also introduced the Python library, Snake‐bite. Snakebite’s client library was explained in detail with multipleexamples. The snakebite CLI was also introduced as a Python alter‐native to the hdfs dfs command.Chapter Summary 13

CHAPTER 2MapReduce with PythonMapReduce is a programming model that enables large volumes ofdata to be processed and generated by dividing work into independ‐ent tasks and executing the tasks in parallel across a cluster ofmachines. The MapReduce programming style was inspired by thefunctional programming constructs map and reduce, which arecommonly used to process lists of data. At a high level, every Map‐Reduce program transforms a list of input data elements into a listof output data elements twice, once in the map phase and once inthe reduce phase.This chapter begins by introducing the MapReduce programmingmodel and describing how data flows through the different phasesof the model. Examples then show how MapReduce jobs can bewritten in Python.Data FlowThe MapReduce framework is composed of three major phases:map, shuffle and sort, and reduce. This section describes each phasein detail.MapThe first phase of a MapReduce application is the map phase.Within the map phase, a function (called the mapper) processes aseries of key-value pairs. The mapper sequentially processes each15

key-value pair individually, producing zero or more output keyvalue pairs (Figure 2-1).Figure 2-1. The mapper is applied to each input key-value pair, pro‐ducing an output key-value pairAs an example, consider a mapper whose purpose is to transformsentences into words. The input to this mapper would be strings thatcontain sentences, and the mapper’s function would be to split thesentences into words and output the words (Figure 2-2).Figure 2-2. The input of the mapper is a string, and the function of themapper is to split the input on spaces; the resulting output is the indi‐vidual words from the mapper’s input16 Chapter 2: MapReduce with Python

Shuffle and SortThe second phase of MapReduce is the shuffle and sort. As the map‐pers begin completing, the intermediate outputs from the mapphase are moved to the reducers. This process of moving outputfrom the mappers to the reducers is known as shuffling.Shuffling is handled by a partition function, known as the parti‐tioner. The partitioner is used to control the flow of key-value pairsfrom mappers to reducers. The partitioner is given the mapper’soutput key and the number of reducers, and returns the index of theintended reducer. The partitioner ensures that all of the values forthe same key are sent to the same reducer. The default partitioner ishash-based. It computes a hash value of the mapper’s output key andassigns a partition based on this result.The final stage before the reducers start processing data is the sort‐ing process. The intermediate keys and values for each partition aresorted by the Hadoop framework before being presented to thereducer.ReduceThe third phase of MapReduce is the reduce phase. Within thereducer phase, an iterator of values is provided to a function knownas the reducer. The iterator of values is a nonunique set of values foreach unique key from the output of the map phase. The reduceraggregates the values for each unique key and produces zero ormore output key-value pairs (Figure 2-3).Data Flow 17

Figure 2-3. The reducer iterates over the input values, producing anoutput key-value pairAs an example, consider a reducer whose purpose is to sum all ofthe values for a key. The input to this reducer is an iterator of all ofthe values for a key, and the reducer sums all of the values. Thereducer then outputs a key-value pair that contains the input keyand the sum of the input key values (Figure 2-4).Figure 2-4. This reducer sums the values for the keys “cat” and “mouse”The next section describes a simple MapReduce application and itsimplementation in Python.Hadoop StreamingHadoop streaming is a utility that comes packaged with the Hadoopdistribution and allows MapReduce jobs to be created with any exe‐cutable as the mapper and/or the reducer. The Hadoop streamingutility enables Python, shell scripts, or any other language to be usedas a mapper, reducer, or both.18 Chapter 2: MapReduce with Python

How It WorksThe mapper and reducer are both executables that read input, lineby line, from the standard input (stdin), and write output to thestandard output (stdout). The Hadoop streaming utility creates aMapReduce job, submits the job to the cluster, and monitors its pro‐gress until it is complete.When the mapper is initialized, each map task launches the specifiedexecutable as a separate process. The mapper reads the input file andpresents each line to the executable via stdin. After the executableprocesses each line of input, the mapper collects the output fromstdout and converts each line to a key-value pair. The key consists ofthe part of the line before the first tab character, and the value con‐sists of the part of the line after the first tab character. If a line con‐tains no tab character, the entire line is considered the key and thevalue is null.When the reducer is initialized, each reduce task launches the speci‐fied executable as a separate process. The reducer converts the inputkey-value pair to lines that are presented to the executable via stdin.The reducer collects the executables result from stdout and convertseach line to a key-value pair. Similar to the mapper, the executablespecifies key-value pairs by separating the key and value by a tabcharacter.A Python ExampleTo demonstrate how the Hadoop streaming utility can run Pythonas a MapReduce application on a Hadoop cluster, the WordCountapplication can be implemented as two Python programs: mapper.pyand reducer.py.mapper.py is the Python program that implements the logic in themap phase of WordCount. It reads data from stdin, splits the linesinto words, and outputs each word with its intermediate count tostdout. The code in Example 2-1 implements the logic in mapper.py.Example 2-1. in/env pythonimport sys# Read each line from stdinHadoop Streaming 19

for line in sys.stdin:# Get the words in each linewords line.split()# Generate the count for each wordfor 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 thereduce phase of WordCount. It reads the results of mapper.py fromstdin, sums the occurrences of each word, and writes the result tostdout. The code in Example 2-2 implements the logic in reducer.py.Example 2-2. bin/env pythonimport syscurr word Nonecurr count 0# Process each key-value pair from the mapperfor line in sys.stdin:# Get the key and value from the current lineword, count line.split('\t')# Convert the count to an intcount int(count)# If the current word is the same as the previous word,# increment its count, otherwise print the words count# to stdoutif word curr word:curr count countelse:# Write word and its number of occurrences as a key-value# pair to stdoutif curr word:print '{0}\t{1}'.format(curr word, curr count)curr word word20 Chapter 2: MapReduce with Python

curr count count# Output the count for the last wordif curr word word:print '{0}\t{1}'.format(curr word, curr count)Before attempting to execute the code, ensure that the mapper.pyand reducer.py files have execution permission. The following com‐mand will enable this for both files: chmod a x mapper.py reducer.pyAlso ensure that the first line of each file contains the proper path toPython. This line enables mapper.py and reducer.py to execute asstandalone executables. The value #!/usr/bin/env python shouldwork for most systems, but if it does not, replace /usr/bin/envpython with the path to the Python execu

Hadoop Distributed File System (HDFS) The Hadoop Distributed File System (HDFS) is a Java-based dis‐ tributed, scalable, and portable filesystem designed to span large clusters of commodity servers. The design of HDFS is based on GFS, the Google File System, which is described in a paper published by Google.