Part 6
In our fifth installment of this series we showed how to implent TF-IDF in Cascading application. If you haven’t read that yet, it’s probably best to start there.
Today's lesson extends the TF-IDF app to show best practices for test-driven development (TDD) at scale. We’ll incorporate unit tests into the build (should have done so sooner), plus show how to leverage TDD features which are unique to Cascading: checkpoints, traps, assertions, etc. These are based on using Checkpoint, Debug, and AssertMatches.
We’ll keep building on this example to show how to leverage “local mode”.
Fortunately, most all of the data required to calculate TF-IDF weight was already available in our Word Count example in Part 4. However, we’ll need to revise the overall workflow, adding more pipe assemblies to it.
AssertionLevel.STRICT as unit test during development, and run against regression data.
As explained quite eloquently in Data Jujitsu by DJ Patil: “It’s impossible to overstress this: 80% of the work in any data project is in cleaning the data ... Work done up front in getting clean data will be amply repaid over the course of the project.”
Checkpoint and Debug are fairly similar – for now. So why use them both? On one hand, Debug can be turned off, e.g., with a command line option. On the other hand, Checkpoint provides an additional feature, since it forces intermediate data to be written out to HDFS. This feature becomes important in the physical plan of the workflow for performance reasons, i.e., to avoid recomputing intermediate results when there are multiple uses downstream. Also, in subsequent releases of Cascading, the Checkpoint class may take on additional meanings.
A conceptual diagram for this implementation of TF-IDF in MapReduce is shown as:
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.
First, let’s add a unit test and show how that works into this example. In the Gradle build script build.gradle we need to modify the compile task to include JUnit and other testing dependencies:
compile( 'cascading:cascading-hadoop:2.0.1' ) { transitive = true }
testCompile( 'org.apache.hadoop:hadoop-test:0.20.2' )
testCompile( 'junit:junit:4.8.+' )
Then we’ll add a test task:
test {
include 'impatient/**'
//makes the standard streams (err and out) visible at console when running tests
testLogging.showStandardStreams = true
//listening to test execution events
beforeTest { descriptor ->
logger.lifecycle("Running test: " + descriptor)
}
onOutput { descriptor, event ->
logger.lifecycle("Test: " + descriptor + " produced standard out/err: " + event.message )
}
}
A little restructuring of the source directories is requried – see our GitHub code repo, where it’s all set up property. Then we add a unit test for our custom function to “scrub” tokens, which was created in Part 3. This goes into a new class ScrubTest.java:
public class ScrubTest
{
@Test
public void testMain() throws Exception
{
ScrubTest tester = new ScrubTest();
Fields fieldDeclaration = new Fields( "doc_id", "token" );
ScrubFunction scrub = new ScrubFunction( fieldDeclaration );
assertEquals( "Scrub", "foo bar", scrub.scrubText( "FoO BAR " ) );
}
}
This is a particularly good place for a unit test. Scrubbing tokens is a likely point at which edge cases get encountered at scale. In practice, you’d probably want to define even more unit tests.
Next, going back to the Main.java module, let’s add sink taps for writing out the checkpointed data plus any trapped data:
String trapPath = args[ 4 ];
String checkPath = args[ 5 ];
Tap trapTap = new Hfs( new TextDelimited( true, "\t" ), trapPath );
Tap checkTap = new Hfs( new TextDelimited( true, "\t" ), checkPath );
Next we’ll modify the head of the existing pipe assembly for TF-IDF to incorporate a Stream Assertion. First we use an AssertMatches to define the expected pattern for input data. Then we apply AssertionLevel.STRICT for validating the data at scale:
// use a stream assertion to validate the input data
Pipe docPipe = new Pipe( "token" );
AssertMatches assertMatches = new AssertMatches( "doc\\d+\\s.*" );
docPipe = new Each( docPipe, AssertionLevel.STRICT, assertMatches );
Next we’ll add a Debug and DebugLevel.VERBOSE to the D branch, to trace the tuple values in the flow there:
// example use of a debug, to observe tuple stream; turn off below
dfPipe = new Each( dfPipe, DebugLevel.VERBOSE, new Debug( true ) );
Next we’ll add a Checkpoint) after the join of the DF and D branches. That forces the tuples at this point in the workflow to be persisted to HDFS:
// create a checkpoint, to observe the intermediate data in DF stream
Checkpoint idfCheck = new Checkpoint( "checkpoint", idfPipe );
Pipe tfidfPipe = new CoGroup( tfPipe, tf_token, idfCheck, df_token );
Next we have a relatively more complex set of taps to connect in the FlowDef, to include output data for TDD-related features:
// connect the taps, pipes, traps, checkpoints, etc., into a flow
FlowDef flowDef = FlowDef.flowDef()
.setName( "tfidf" )
.addSource( docPipe, docTap )
.addSource( stopPipe, stopTap )
.addTailSink( tfidfPipe, tfidfTap )
.addTailSink( wcPipe, wcTap )
.addTrap( docPipe, trapTap )
.addCheckpoint( idfCheck, checkTap );
Last, we’ll specify the verbosity level for the debug trace, and the strictness level for the stream assertion:
// set to DebugLevel.VERBOSE for trace, or DebugLevel.NONE in production
flowDef.setDebugLevel( DebugLevel.VERBOSE );
// set to AssertionLevel.STRICT for all assertions, or AssertionLevel.NONE in production
flowDef.setAssertionLevel( AssertionLevel.STRICT );
Modify the Main method to make those changes, then build a JAR file. You should be good to go. For those keeping score, the resulting physical plan in MapReduce for Part 5 now uses eleven mappers and nine reducers. That amount jumped by 5x since our previous example.
The diagram for the Cascading flow will be in the dot/ subdirectory after the app runs. Here we have annotated it to show where the mapper and reducer phases are running, and also the sections which were added since Part 5:
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 data/en.stop output/tfidf output/trap output/check
Output text gets stored in the partition file output/tfidf which you can then verify:
more output/tfidf/part-00000
more output/trap/part-m-00001-00000
more output/check/part-00000
Here’s a log file from our run of the sample app, part 5. 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.