Cascading Multitool

As explained above Multitool supports all Cascading platforms. In order for Multitool to work, you have to tell it which platform you want to use. Below we see all ways of setting the platform in order of precedence:

The first way of setting the platform is using the --platform <name> command line switch. Valid platform names are hadoop, hadoop2-mr1 and local. The next possibilty is setting the MULTITOOL_PLATFORM environment variable. If that is not set, Multitool will try to use the CASCADING_PLATFORM environment variable. If that is also not set, it will look for a file called .cascading in the current working directory. The .cascading file is a Java properties file and can be used to set the platform via the keys multitool.platform.name or cascading.platform.name. If no file is present in the current working directory, multitool will look for a .cascading file in the user’s $HOME directory, which may contain the same keys.

The Multitool shell wrapper will chose hadoop as the default platform, if none of the above settings can be found.

The preferred way of running Multitool is using the shell wrapper coming with the distribution:

It is also possible to download the hadoop and/or hadoop2 jar files compatible with your Hadoop version. This is considered an expert mode of using Multitool. If you are unsure, stick with the full client install.

If you choose to work with the jar files directly, you can download them like so:

Hadoop 1:

To run it do this:

For Hadoop 2:

To run it do this:

All downloads are hosted on a public S3 bucket, which makes it easy to deploy Multitool in a bootstrap action an Amazon EMR cluster.

Consider having a file called days.txt which contains the following text:

One could use the grep utility to find text lines in that file which match a given pattern. For example, to write all the lines containing the string Tuesday into a file called output/tuesday.txt:

The results in the output/tuesday.txt file would be:

To use Multitool for the exact same work:

Note that Multitool uses Cascading operations for its parameters - a source for input and a sink for output in this case. That allows for a graph of MapReduce jobs to be connected together as a workflow, run as a single application.

Everybody loves the ubiquitous MapReduce example of Word Count, so let’s take a look at it in terms of Multitool compared with Linux command line utilities. First, the Multitool version:

Note that you may need to escape some characters when running from a Linux shell, such as in the expr statement above. That may not be needed in other Hadoop environments, such as Elastic MapReduce.

Log output from an example run is stored in a gist.

The first 10 lines in the results would be:

The same work performed with Linux command line utilities would be:

These two different approaches use roughly the same length of command lines. However, the Multitool approach is stated in terms of set operations used in database theory, and is much closer conceptually to what’s happening in MapReduce. When you need to think in terms of data processing with MapReduce, then Multitool provides a simpler approach. It also scales as big as your Hadoop cluster.

We can look at this differently by adding another Multitool parameter --dot=flow.dot to generate a file flow.dot, which depicts the MapReduce workflow as a directed acyclic graph or DAG.

The generated graph (in the center column of the following diagram) shows how this workflow is built out of two MapReduce jobs connected in a flow:

The first mapper phase loads from days.txt and processes the text. It splits each line on the spaces, converts the words to lowercase, then emits those words as tuples for the following reducer. The first reducer phase performs a group by to count the occurrence of each word. The second mapper phase simply loads this input (an identity mapper), and the second reducer phase sorts the results in decreasing order based on count.

Back to the point about learning to program MapReduce efficiently, let’s examine each of the Multitool commands which were used:

The source argument specifies that the input should be read from the file days.txt for the initial map phase. That is called a tap in Cascading.

The expr argument applies a function to the data, in this case a Java String function to convert the data to lower case.

The two gen arguments split the input text using a space delimiter, where the resulting stream of words is used for the input tuples to the rest of the flow.

The next two arguments, group=0 and count specify that fields emitted by the regular expression for the preceding gen should be grouped and counted according to the first column, i.e. count the number of occurrences of each word in the stream. This will produce two columns: words and counts. Note that a group must be used before a count argument.

The next set of group arguments cause the output to be sorted in descending order, based on the counts.

Finally, the two sink arguments write the output to the “output/” directory, where “sink.part=1” means that all output should be consolidated into a single file.

Next, let’s join a view of the data in days.txt with a view of the data in another file called rhymes.txt which contains the following text:

For this example, we take two flows, one based on days.txt which we label as lhs and another based on rhymes.text which we will label as “rhs”, apply “unique” to both, and then join them to produce the output in a directory output/join/:

The results provide the intersection of words common to both input files:

Since this workflow is a graph and not a linear pipeline, this example becomes more complex to state as a pipe of Unix/Linux command line utilities. We could use temporary files, and multiple command lines. However, keep in mind that the Multitool example only had two flows being joined, while it could have many flows joined in a complex DAG. That’s where the power of Cascading really begins to shine.