When Google was launched in 1998, the amount of online data was already enormous with approximately 2.4 million websites²⁰—truly big data. Today there are now nearly two billion websites²¹ (almost a thousandfold increase) and Google is handling over two trillion searches per year!²² Having used Google search since its inception, our sense is that today’s responses are significantly faster.

When Google was developing their search engine, they knew that they needed to return search results quickly. The only practical way to do this was to store and index the entire Internet using a clever combination of secondary storage and main memory. Computers of that time couldn’t hold that amount of data and could not analyze that amount of data fast enough to guarantee prompt search-query responses. So Google developed a clustering system, tying together vast numbers of computers—called nodes. Because having more computers and more connections between them meant greater chance of hardware failures, they also built in high levels of redundancy to ensure that the system would continue functioning even if nodes within clusters failed. The data was distributed across all these inexpensive “commodity computers.” To satisfy a search request, all the computers in the cluster searched in parallel the portion of the web they had locally. Then the results of those searches were gathered up and reported back to the user.

To accomplish this, Google needed to develop the clustering hardware and software, including distributed storage. Google publishes its designs, but did not open source its software. Programmers at Yahoo!, working from Google’s designs in the “Google File System” paper,²³ then built their own system. They open-sourced their work and the Apache organization implemented the system as Hadoop. The name came from an elephant stuffed animal that belonged to a child of one of Hadoop’s creators.

Two additional Google papers also contributed to the evolution of Hadoop—“MapReduce: Simplified Data Processing on Large Clusters”²⁴ and “Bigtable: A Distributed Storage System for Structured Data,”²⁵ which was the basis for Apache HBase (a NoSQL key–value and column-based database).²⁶

In the next several subsections, you’ll create a cloud-based, multi-node cluster of computers using Microsoft Azure HDInsight. Then, you’ll use the service’s capabilities to demonstrate Hadoop MapReduce running on that cluster. The MapReduce task you’ll define will determine the length of each word in RomeoAndJuliet.txt (from the “Natural Language Processing” chapter), then summarize how many words of each length there are. After defining the task’s mapping and reduction steps, you’ll submit the task to your HDInsight cluster, and Hadoop will decide how to use the cluster of computers to perform the task.

Most major cloud vendors have support for Hadoop and Spark computing clusters that you can configure to meet your application’s requirements. Multi-node cloud-based clusters typically are paid services, though most vendors provide free trials or credits so you can try out their services.

We want you to experience the process of setting up clusters and using them to perform tasks. So, in this Hadoop example, you’ll use Microsoft Azure’s HDInsight service to create cloud-based clusters of computers in which to test our examples. Go to

https://azure.microsoft.com/en-us/free 

to sign up for an account. Microsoft requires a credit card for identity verification.

Various services are always free and some you can continue to use for 12 months. For information on these services see:

https://azure.microsoft.com/en-us/free/free-account-faq/

Microsoft also gives you a credit to experiment with their paid services, such as their HDInsight Hadoop and Spark services. Once your credits run out or 30 days pass (whichever comes first), you cannot continue using paid services unless you authorize Microsoft to charge your card.

Because you’ll use your new Azure account’s credit for these examples,³⁵ we’ll discuss how to configure a low-cost cluster that uses less computing resources than Microsoft allocates by default.³⁶ Caution: Once you allocate a cluster, it incurs costs whether you’re using it or not. So, when you complete this case study, be sure to delete your cluster(s) and other resources, so you don’t incur additional charges. For more information, see:

https://docs.microsoft.com/en-us/azure/azure-resource-manager/resource-group-portal

For Azure-related documentation and videos, visit:

• https://docs.microsoft.com/en-us/azure/—the Azure documentation.

• https://channel9.msdn.com/—Microsoft’s Channel 9 video network.

• https://www.youtube.com/user/windowsazure—Microsoft’s Azure channel on YouTube.

https://docs.microsoft.com/en-us/azure/hdinsight/hadoop/apache-hadoop-linux-create-cluster-get-started-portal

https://portal.azure.com

https://azure.microsoft.com/en-us/resources/knowledge-center/technical-chat/

For languages like Python that are not natively supported in Hadoop, you must use Hadoop streaming to implement your tasks. In Hadoop streaming, the Python scripts that implement the mapping and reduction steps use the standard input stream and standard output stream to communicate with Hadoop. Usually, the standard input stream reads from the keyboard and the standard output stream writes to the command line. However, these can be redirected (as Hadoop does) to read from other sources and write to other destinations. Hadoop uses the streams as follows:

• Hadoop supplies the input to the mapping script—called the mapper. This script reads its input from the standard input stream.

• The mapper writes its results to the standard output stream.

• Hadoop supplies the mapper’s output as the input to the reduction script—called the reducer—which reads from the standard input stream.

• The reducer writes its results to the standard output stream.

• Hadoop writes the reducer’s output to the Hadoop file system (HDFS).

The mapper and reducer terminology used above should sound familiar to you from our discussions of functional-style programming and filter, map and reduce in the “Sequences: Lists and Tuples” chapter.

In this section, you’ll create a mapper script that takes lines of text as input from Hadoop and maps them to key–value pairs in which each key is a word, and its corresponding value is 1. The mapper sees each word individually so, as far as it is concerned, there’s only one of each word. In the next section, the reducer will summarize these key–value pairs by key, reducing the counts to a single count for each key. By default, Hadoop expects the mapper’s output and the reducer’s input and output to be in the form of key–value pairs separated by a tab.

In the mapper script (length_mapper.py), the notation #! in line 1 tells Hadoop to execute the Python code using python3, rather than the default Python 2 installation. This line must come before all other comments and code in the file. At the time of this writing, Python 2.7.12 and Python 3.5.2 were installed. Note that because the cluster does not have Python 3.6 or higher, you cannot use f-strings in your code.

1 #!/usr/bin/env python3
2 # length_mapper.py
3 """Maps lines of text to key-value pairs of word lengths and 1."""
4 import sys
5
6 def tokenize_input():
7     """Split each line of standard input into a list of strings."""
8     for line in sys.stdin:
9         yield line.split()
10
11 # read each line in the the standard input and for every word
12 # produce a key-value pair containing the word, a tab and 1
13 for line in tokenize_input():
14     for word in line:
15         print(str(len(word)) + '	1')

Generator function tokenize_input (lines 6–9) reads lines of text from the standard input stream and for each returns a list of strings. For this example, we are not removing punctuation or stop words as we did in the “Natural Language Processing” chapter.

When Hadoop executes the script, lines 13–15 iterate through the lists of strings from tokenize_input. For each list (line) and for every string (word) in that list, line 15 outputs a key–value pair with the word’s length as the key, a tab ( ) and the value 1, indicating that there is one word (so far) of that length. Of course, there probably are many words of that length. The MapReduce algorithm’s reduction step will summarize these key–value pairs, reducing all those with the same key to a single key–value pair with the total count.

In the reducer script (length_reducer.py), function tokenize_input (lines 8–11) is a generator function that reads and splits the key–value pairs produced by the mapper. Again, the MapReduce algorithm supplies the standard input. For each line, tokenize_input strips any leading or trailing whitespace (such as the terminating newline) and yields a list containing the key and a value.

1 #!/usr/bin/env python3
2 # length_reducer.py
3 """Counts the number of words with each length."""
4 import sys
5 from itertools import groupby
6 from operator import itemgetter
7
8 def tokenize_input():
9     """Split each line of standard input into a key and a value."""
10    for line in sys.stdin:
11       yield line.strip().split('	')
12
13 # produce key-value pairs of word lengths and counts separated by tabs
14 for word_length, group in groupby(tokenize_input(), itemgetter(0)):
15     try:
16         total = sum(int(count) for word_length, count in group)
17         print(word_length + '	' + str(total))
18     except ValueError:
19         pass # ignore word if its count was not an integer

When the MapReduce algorithm executes this reducer, lines 14–19 use the groupby function from the itertools module to group all word lengths of the same value:

• The first argument calls tokenize_input to get the lists representing the key–value pairs.

• The second argument indicates that the key–value pairs should be grouped based on the element at index 0 in each list—that is the key.

Line 16 totals all the counts for a given key. Line 17 outputs a new key–value pair consisting of the word and its total. The MapReduce algorithm takes all the final word-count outputs and writes them to a file in HDFS—the Hadoop file system.

Next, you’ll upload files to the cluster so you can execute the example. In a Command Prompt, Terminal or shell, change to the folder containing your mapper and reducer scripts and the RomeoAndJuliet.txt file. We assume all three are in this chapter’s ch17 examples folder, so be sure to copy your RomeoAndJuliet.txt file to this folder first.

scp length_mapper.py length_reducer.py RomeoAndJuliet.txt
     sshuser@YourClusterName-ssh.azurehdinsight.net:

ssh sshuser@YourClusterName-ssh.azurehdinsight.net

hadoop fs -copyFromLocal RomeoAndJuliet.txt
     /example/data/RomeoAndJuliet.txt

Now you can run the MapReduce job for RomeoAndJuliet.txt on your cluster by executing the following command. For your convenience, we provided the text of this command in the file yarn.txt with this example, so you can copy and paste it. We reformatted the command here for readability:

yarn jar /usr/hdp/current/hadoop-mapreduce-client/hadoop-streaming.jar
   -D mapred.output.key.comparator.class=
      org.apache.hadoop.mapred.lib.KeyFieldBasedComparator
   -D mapred.text.key.comparator.options=-n
   -files length_mapper.py,length_reducer.py
   -mapper length_mapper.py
   -reducer length_reducer.py
   -input /example/data/RomeoAndJuliet.txt
   -output /example/wordlengthsoutput

The yarn command invokes the Hadoop’s YARN (“yet another resource negotiator”) tool to manage and coordinate access to the Hadoop resources the MapReduce task uses. The file hadoop-streaming.jar contains the Hadoop streaming utility that allows you to use Python to implement the mapper and reducer. The two -D options set Hadoop properties that enable it to sort the final key–value pairs by key (KeyFieldBasedComparator) in descending order numerically (-n; the minus indicates descending order) rather than alphabetically. The other command-line arguments are:

• -files—A comma-separated list of file names. Hadoop copies these files to every node in the cluster so they can be executed locally on each node.

• -mapper—The name of the mapper’s script file.

• -reducer—The name of the reducer’s script file

• -input—The file or directory of files to supply as input to the mapper.

• -output—The HDFS directory in which the output will be written. If this folder already exists, an error will occur.

The following output shows some of the feedback that Hadoop produces as the MapReduce job executes. We replaced chunks of the output with … to save space and bolded several lines of interest including:

• The total number of “input paths to process”—the 1 source of input in this example is the RomeoAndJuliet.txt file.

• The “number of splits”—2 in this example, based on the number of worker nodes in our cluster.

• The percentage completion information.

• File System Counters, which include the numbers of bytes read and written.

• Job Counters, which show the number of mapping and reduction tasks used and various timing information.

• Map-Reduce Framework, which shows various information about the steps performed.

packageJobJar: [] [/usr/hdp/2.6.5.3004-13/hadoop-
mapreduce/hadoopstreaming-2.7.3.2.6.5.3004-13.jar] /tmp/streamjob2764990629848702405.jar
tmpDir=null
...
18/12/05 16:46:25 INFO mapred.FileInputFormat: Total input paths to
process : 1
18/12/05 16:46:26 INFO mapreduce.JobSubmitter: number of splits:2
...
18/12/05 16:46:26 INFO mapreduce.Job: The url to track the job: http://
hn0-paulte.y3nghy5db2kehav5m0opqrjxcb.cx.internal.cloudapp.net:8088
/proxy/application_1543953844228_0025/
...
18/12/05 16:46:35 INFO mapreduce.Job: map 0% reduce 0%
18/12/05 16:46:43 INFO mapreduce.Job: map 50% reduce 0%
18/12/05 16:46:44 INFO mapreduce.Job: map 100% reduce 0%
18/12/05 16:46:48 INFO mapreduce.Job: map 100% reduce 100%
18/12/05 16:46:50 INFO mapreduce.Job: Job job_1543953844228_0025
completed successfully
18/12/05 16:46:50 INFO mapreduce.Job: Counters: 49
         File System Counters
              FILE: Number of bytes read=156411
              FILE: Number of bytes written=813764
...
         Job Counters
              Launched map tasks=2
              Launched reduce tasks=1
...
         Map-Reduce Framework
              Map input records=5260
              Map output records=25956
              Map output bytes=104493
              Map output materialized bytes=156417
              Input split bytes=346
              Combine input records=0
              Combine output records=0
              Reduce input groups=19
              Reduce shuffle bytes=156417
              Reduce input records=25956
              Reduce output records=19
              Spilled Records=51912
              Shuffled Maps =2
              Failed Shuffles=0
              Merged Map outputs=2
              GC time elapsed (ms)=193
              CPU time spent (ms)=4440
              Physical memory (bytes) snapshot=1942798336
              Virtual memory (bytes) snapshot=8463282176
              Total committed heap usage (bytes)=3177185280
...
18/12/05 16:46:50 INFO streaming.StreamJob: Output directory: /example/
wordlengthsoutput

hdfs dfs -text /example/wordlengthsoutput/part-00000

18/12/05 16:47:19 INFO lzo.GPLNativeCodeLoader: Loaded native gpl library
18/12/05 16:47:19 INFO lzo.LzoCodec: Successfully loaded & initialized
native-lzo library [hadoop-lzo rev
b5efb3e531bc1558201462b8ab15bb412ffa6b89]
1        1140
2        3869
3        4699
4        5651
5        3668
6        2719
7        1624
8        1062
9        855
10       317
11       189
12       95
13       35
14       13
15       9
16       6
17       3
18       1
23       1

https://docs.microsoft.com/en-us/azure/azure-resource-manager/resource-group-portal