Part 2
In our first installment of this series we showed how to create the simplest possible Cascading 2.0 application. If you haven't read that yet, it's probably best to start there.
Today's lesson takes the same app and stretches it a bit further. Undoubtedy you've seen Word Count before. We'd feel remiss at Cascading if we did not provide a Word Count example. It's the "Hello World" of MapReduce apps. Fortunately, this code is one of the basic steps toward developing a TF-IDF implementation. How convenient. We'll also show how to use Cascading to generate a visualization of your MapReduce app.
Download source for this example on GitHub. For quick reference, the source code and a log for this example is listed in a gist. The input data stays the same as in the earlier code
Note that the names of the taps have changed. Instead of inTap and outTap, we're now using docTap and wcTap. We'll be adding more taps, so this will help to have more descriptive names. Makes it simpler to follow all the plumbing.
Previously we defined a simple pipe to connect the taps. This example shows a more complex pipe. We use a generator inside of an each, to split the document text into a token stream. We use a regex to split on word boundaries:
Fields token = new Fields( "token" );
Fields text = new Fields( "text" );
RegexSplitGenerator splitter = new RegexSplitGenerator( token, regex_string );
Fields fieldDeclaration = new Fields( "doc_id", "token" );
Pipe docPipe = new Each( "token", text, splitter, fieldDeclaration );
Unfortunately, markdown on the GitHub wiki seems to turn psychotic when it parses that regex string value, even with escapes. See the gist or the src code repo.
Out of that pipe, we'll get a tuple stream of doc_id and token, to feed into the count. You can change the regex to handle more complex cases of splitting tokens -- without having to rewrite a different generator class.
Next, we define a stream which just has the tokens, using a retain operator to discard all other fields:
Pipe wcPipe = new Pipe( "wc", docPipe );
wcPipe = new Retain( wcPipe, token );
Then we use a groupby to count the occurrences of each token:
wcPipe = new GroupBy( wcPipe, token );
wcPipe = new Every( wcPipe, Fields.ALL, new Count(), Fields.ALL );
From that pipe, we'll have a resulting tuple stream of token and count for the output. So we connect up the plumbing with a flowdef:
FlowDef flowDef = FlowDef.flowDef().setName( "wc" );
flowDef.addSource( docPipe, docTap );
flowDef.addTailSink( wcPipe, wcTap );
Finally, we generate a DOT file, to depict the Cascading flow graphically. You can load the DOT file into OmniGraffle or Visio. Those diagrams are really helpful for troubleshooting MapReduce workflows in Cascading:
Flow wcFlow = flowConnector.connect( flowDef );
wcFlow.writeDOT( "dot/wc.dot" );
wcFlow.complete();
Place those source lines all into a Main method, then build a JAR file. You should be good to go.
Here we have taken the diagram for the Cascading flow, and annotated it to show where the mapper and reducer phases are running:
If you want to read in more detail about the classes in the Cascading API which were used, see the Cascading 2.0 User Guide and JavaDoc.
The build for this example is based on using Gradle. The script is in build.gradle and to generate an IntelliJ project use:
gradle ideaModule
To build the sample app from the command line use:
gradle clean jar
What you should have at this point is a JAR file which is nearly ready to drop into your Maven repo -- almost. Actually, we provide a community jar repository for Cascading libraries and extensions at http://conjars.org
Before running this sample app, you'll need to have a supported release of Apache Hadoop installed. Here's what was used to develop and test our example code:
$ hadoop version
Hadoop 1.0.3
Be sure to set your HADOOP_HOME environment variable. Then clear the output directory (Apache Hadoop insists, if you're running in standalone mode) and run the app:
rm -rf output
hadoop jar ./build/libs/impatient.jar data/rain.txt output/wc
Output text gets stored in the partition file output/wc which you can then verify:
more output/wc/part-00000
Again, here's a log file from our run of the sample app, part 2. If your run looks terribly different, something is probably not set up correctly. Drop us a line on the cascading-user email forum. Or visit one of our user group meetings. [Coming up real soon...]
Stay tuned for the next installments of our Cascading for the Impatient series.