Cascading - User Guide

Cascading is an API for defining, sharing, and executing data processing workflows on a distributed data grid or cluster.

Cascading relies on Apache Hadoop. To use Cascading, Hadoop must be installed locally for development and testing, and a Hadoop cluster must be deployed for production applications.

Cascading greatly simplifies the complexities with Hadoop application development, job creation, and job scheduling.

Cascading was developed to allow organizations to rapidly develop complex data processing applications. These applications come in two extremes.

On one hand, there is too much data for a single computing system to manage effectively. Developers have decided to adopt Apache Hadoop as the base computing infrastructure, but realize that developing reasonably useful applications on Hadoop is not trivial. Cascading eases the burden on developers by allowing them to rapidly create, refactor, test, and execute complex applications that scale linearly across a cluster of computers.

On the other hand, managers and developers realize the complexity of the processes in their data canter is getting out of hand with one-off data-processing applications living wherever there is enough disk space. Subsequently they have decided to adopt Apache Hadoop to gain access to its "Global Namespace" file system which allows for a single reliable storage framework. Cascading eases the learning curve for developers to convert their existing applications for execution on a Hadoop cluster. It further allows for developers to create reusable libraries and application for use by analysts who need to extract data from the Hadoop file system.

Cascading was designed to support three user roles. The application Executor, process Assembler, and the operation Developer.

The application Executor is someone, a developer or analyst, or some system (like a cron job) which runs a data processing application on a given cluster. This is typically done via the command line using a pre-packaged Java Jar file compiled against the Apache Hadoop and Cascading libraries. This application may accept parameters to customize it for an given execution and generally results in a set of data the user will export from the Hadoop file system for some specific purpose.

The process Assembler is someone who assembles data processing workflows into unique applications. This is generally a development task of chaining together operations that act on input data sets to produce one or more output data sets. This task can be done using the raw Java Cascading API or via a scripting language like Groovy, JRuby, or Jython.

The operation Developer is someone who writes individual functions or operations, typically in Java, or reusable sub-assemblies that act on the data that pass through the data processing workflow. A simple example would be a parser that takes a string and converts it to an Integer. Operations are equivalent to Java functions in the sense that they take input arguments and return data. And they can execute at any granularity, simply parsing a string, or performing some complex routine on the argument data using third-party libraries.

All three roles can be a developer, but the API allows for a clean separation of responsibilities for larger organizations that need non-developers to run ad-hoc applications or build production processes on a Hadoop cluster.

From the Hadoop website, it “is a software platform that lets one easily write and run applications that process vast amounts of data”.

To be a little more specific, Hadoop provides a storage layer that holds vast amounts of data, and an execution layer for running an application in parallel across the cluster against parts of the stored data.

The storage layer, the Hadoop File System (HDFS), looks like a single storage volume that has been optimized for many concurrent serialized reads of large data files. Where "large" ranges from Gigabytes to Petabytes. But it only supports a single writer. Thus random access to the data is not really possible in an efficient manner. But this is why it is so performant and reliable. Reliable in part because this restriction allows for the data to be replicated across the cluster reducing the chance of data loss.

The execution layer relies on a "divide and conquer" strategy called MapReduce. MapReduce is beyond the scope of this document, but suffice it to say, it can be so difficult to develop "real world" applications against that Cascading was created to offset the complexity.

Apache Hadoop is an Open Source Apache project and is freely available. It can be downloaded from here the Hadoop website, http://hadoop.apache.org/core/.

The Cascading processing model is based on a "pipes and filters" metaphor. The developer uses the Cascading API to assemble pipelines that split, merge, group, or join streams of data while applying operations to each data record or groups of records.

In Cascading, we call a data record a Tuple, a pipeline a pipe assembly, and a series of Tuples passing through a pipe assembly is called a tuple stream.

Pipe assemblies are assembled independently from what data they will process. Before a pipe assembly can be executed, it must be bound to data sources and data sinks, called Taps. The process of binding pipe assemblies to sources and sinks results in a Flow. Flows can be executed on a data cluster like Hadoop.

Finally, many Flows can be grouped together and executed as a single process. If one Flow depends on the output of another Flow, it will not be executed until all its data dependencies are satisfied. This collection of Flows is called a Cascade.

Pipe assemblies define what work should be done against a tuple stream, where during runtime tuple streams are read from Tap sources and are written to Tap sinks. Pipe assemblies may have multiple sources and multiple sinks and they can define splits, merges, and joins to manipulate how the tuple streams interact.

There are only five Pipe types: Pipe, Each, GroupBy, CoGroup, Every, and SubAssembly.

Assembling Pipe Assemblies

Pipe assemblies are created by chaining cascading.pipe.Pipe classes and Pipe subclasses together. Chaining is accomplished by passing previous Pipe instances to the constructor of the next Pipe instance.

Example 3.1. Chaining Pipes

// the "left hand side" assembly head
Pipe lhs = new Pipe( "lhs" );

lhs = new Each( lhs, new SomeFunction() );
lhs = new Each( lhs, new SomeFilter() );

// the "right hand side" assembly head
Pipe rhs = new Pipe( "rhs" );

rhs = new Each( rhs, new SomeFunction() );

// joins the lhs and rhs
Pipe join = new CoGroup( lhs, rhs );

join = new Every( join, new SomeAggregator() );

join = new GroupBy( join );

join = new Every( join, new SomeAggregator() );

// the tail of the assembly
join = new Each( join, new SomeFunction() );

The above example, if visualized, would look like the diagram below.

Besides defining the paths tuple streams take through splits, merges, grouping, and joining, pipe assemblies transform or filter the stored values in each Tuple. This is accomplished by applying an Operation to each Tuple or group of Tuples as the tuple stream passes through the pipe assembly. To do that, the values in the Tuple typically are given field names, in the same fashion columns are named in a database so that they can be referenced or selected.

Operation

Operations (cascading.operation.Operation) accept an input argument Tuple, and output zero or more result Tuples. There are a few sub-types of operations defined below. Cascading has a number of generic Operations that can be reused, or developers can create their own custom Operations.

Tuple

In Cascading, we call each record of data a Tuple (cascading.tuple.Tuple), and a series of Tuples are a tuple stream. Think of a Tuple as an Array of values where each value can be any java.lang.Comparable Java type.

Fields

Fields (cascading.tuple.Fields) either declare the field names in a Tuple. Or reference values in a Tuple as a selector. Fields can either be string names ("first_name"), integer positions (-1 for the last value), or a substitution ( Fields.ALL to select all values in the Tuple, like an asterisk (*) in SQL).

Each and Every Pipes

The Each and Every pipe types are the only pipes that can be used to apply Operations to the tuple stream.

The Each pipe applies an Operation to "each" Tuple as it passes through the pipe assembly. The Every pipe applies an Operation to "every" group of Tuples as they pass through the pipe assembly, on the tail end of a GroupBy or CoGroup pipe.

new Each( previousPipe, argumentSelector, operation, outputSelector )

new Every( previousPipe, argumentSelector, operation, outputSelector )

Both the Each and Every pipe take a Pipe instance, an argument selector, Operation instance, and a output selector on the constructor. Where each selector is a Fields instance.

The Each pipe may only apply Functions and Filters to the tuple stream as these operations may only operate on one Tuple at a time.

The Every pipe may only apply Aggregators and Buffers to the tuple stream as these operations may only operate on groups of tuples, one grouping at a time.

The argument selector selects values from the input Tuple to be passed to the Operation as argument values. The output selector selects the output Tuple from an "appended" version of the input Tuple and the Operation result Tuple. The output Tuple becomes the input Tuple to the next pipe in the pipe assembly. If a Function emits more than one Tuple, this process will be repeated for each result Tuple against the original input Tuple, depending on the output selector, input Tuple values could be duplicated across each output Tuple.

If the argument selector is not given, the whole input Tuple (Fields.ALL) is passed to the Operation as argument values. If the result selector is not given on an Each pipe, the Operation results are returned by default (Fields.RESULTS), replacing the input Tuple values in the tuple stream. This really only applies to Functions, as Filters either discard the input Tuple, or return the input Tuple intact. There is no opportunity to provide an output selector.

For the Every pipe, the Aggregator results are appended to the input Tuple (Fields.ALL) by default.

It is important to note that the Every pipe associates Aggregator results with the current group Tuple. For example, if you are grouping on the field "department" and counting the number of "names" grouped by that department, the output Fields would be ["department","num_employees"]. This is true for both Aggregator, seen above, and Buffer.

If you were also adding up the salaries associated with each "name" in each "department", the output Fields would be ["department","num_employees","total_salaries"]. This is only true for chains of Aggregator Operations, you may not chain Buffer Operations.

For the Every pipe when used with a Buffer the behavior is slightly different. Instead of associating the Buffer results with the current grouing Tuple, they are associated with the current values Tuple, just like an Each pipe does with a Function. This might be slightly more confusing, but provides much more flexibility.

GroupBy and CoGroup Pipes

The GroupBy and CoGroup pipes serve two roles. First, they emit sorted grouped tuple streams allowing for Operations to be applied to sets of related Tuple instances. Where "sorted" means the tuple groups are emitted from the GroupBy and CoGroup pipes in sort order of the field values the groups were grouped on.

Second, they allow for two streams to be either merged or joined. Where merging allows for two or more tuple streams originating from different sources to be treated as a single stream. And joining allows two or more streams to be "joined" (in the SQL sense) on a common key or set of Tuple values in a Tuple.

It is not required that an Every follow either GroupBy or CoGroup, an Each may follow immediately after. But an Every many not follow an Each.

It is important to note, for both GroupBy and CoGroup, the values being grouped on must be the same type. If your application attempts to GroupBy on the field "size", but the value alternates between a String and a Long, Hadoop will fail internally attempting to apply a Java Comparator to them. This also holds true for the sort-by fields in GroupBy.

GroupBy accepts one or more tuple streams. If two or more, they must all have the same field names.

Example 3.2. Grouping a Tuple Stream

Pipe merge = new GroupBy( assembly, new Fields( "group1", "group2" ) );

The example above simply creates a new tuple stream where Tuples with the same values in "group1" and "group2" can be processed as a set by an Aggregator or Buffer Operation. The resulting stream of tuples will be sorted by the values in "group1" and "group2".

Example 3.3. Merging a Tuple Stream

Pipe[] pipes = Pipe.pipes( lhs, rhs );
Pipe merge = new GroupBy( pipes, new Fields( "group1", "group2" ) );

This example merges two streams ("lhs" and "rhs") into one tuple stream and groups the resulting stream on the fields "group1" and "group2", in the same fashion as the previous example.

CoGroup accepts two or more tuple streams and does not require any common field names. The grouping fields must be provided for each tuple stream.

Example 3.4. Joining a Tuple Stream

Fields lhsFields = new Fields( "fieldA", "fieldB" );
Fields rhsFields = new Fields( "fieldC", "fieldD" );
Pipe merge = new CoGroup( lhs, lhsFields, rhs, rhsFields, new InnerJoin() );

This example joins two streams ("lhs" and "rhs") on common values. Note that common field names are not required here. Actually, if there were any common field names, the Cascading planner would throw an error as duplicate field names are not allowed.

This is significant because of the nature of joining streams.

The first stage of joining has to do with identifying field names that represent the grouping key for a given stream. The second stage is emitting a new Tuple with the joined values, this includes the grouping values, and the other values.

In the above example, we see what "logically" happens during a join. Here we join two streams on the "url" field which happens to be common to both streams. The result is simply two Tuple instances with the same "url" appended together into a new Tuple. In practice this would fail since the result Tuple has duplicate field names. The CoGroup pipe has the declaredFields argument allowing the developer to declare new unique field names for the resulting tuple.

Example 3.5. Joining a Tuple Stream with Duplicate Fields

Fields common = new Fields( "url" );
Fields declared = new Fields( "url1", "word", "wd_count", "url2", "sentence", "snt_count" );
Pipe merge = new CoGroup( lhs, common, rhs, common, declared, new InnerJoin() );

Here we see an example of what the developer could have named the fields so the planner would not fail.

It is important to note that Cascading could just magically create a new Tuple by removing the duplicate grouping fields names so the user isn't left renaming them. In the above example, the duplicate "url"columns could be collapsed into one, as they are the same value. This is not done because field names are a user convenience, the primary mechanism to manipulate Tuples is through positions, not field names. So the result of every Pipe (Each, Every, CoGroup, GroupBy) needs to be deterministic. This gives Cascading a big performance boost, provides a means for sub-assemblies to be built without coupling to any "domain level" concepts (like "first_name", or "url), and allows for higher level abstractions to be built on-top of Cascading simply.

In the example above, we explicitly set a Joiner class to join our data. The reason CoGroup is named "CoGroup" and not "Join" is because joining data is done after all the parallel streams are co-grouped by their common keys. The details are not terribly important, but note that a "bag" of data for every input tuple stream must be created before an join operation can be performed. Each bag consists of all the Tuple instances associated with a given grouping Tuple.

Above we see two bags, one for each tuple stream ("lhs" and "rhs"). Each Tuple in bag is independent but all Tuples in both bags have the same "url" value since we are grouping on "url", from the previous example. A Joiner will match up every Tuple on the "lhs" with a Tuple on the "rhs". An InnerJoin is the most common. This is where each Tuple on the "lhs" is matched with every Tuple on the "rhs". This is the default behaviour one would see in SQL when doing a join. If one of the bags was empty, no Tuples would be joined. An OuterJoin allows for either bag to be empty, and if that is the case, a Tuple full of null values would be substituted.

Above we see all supported Joiner types.

LHS = [0,a] [1,b] [2,c]
RHS = [0,A] [2,C] [3,D]

Using the above simple data sets, we will define each join type where the values are joined on the first position, a numeric value. Note when Cascading joins Tuples, the resulting Tuple will contain all the incoming values. The duplicate common key(s) is not discarded if given. And on outer joins, where there is no equivalent key in the alternate stream, null values are used as placeholders.

InnerJoin

An Inner join will only return a joined Tuple if neither bag has is empty.

[0,a,0,A] [2,c,2,C]

OuterJoin

An Outer join will join if either the left or right bag is empty.

[0,a,0,A] [1,b,null,null] [2,c,2,C] [null,null,3,D]

LeftJoin

A Left join can also be stated as a Left Inner and Right Outer join, where it is fine if the right bag is empty.

[0,a,0,A] [1,b,null,null] [2,c,2,C]

RightJoin

A Right join can also be stated as a Left Outer and Right Inner join, where it is fine if the left bag is empty.

[0,a,0,A] [2,c,2,C] [null,null,3,D]

MixedJoin

A Mixed join is where 3 or more tuple streams are joined, and each pair must be joined differently. See the cascading.pipe.cogroup.MixedJoin class for more details.

Custom

A custom join is where the developer subclasses the cascading.pipe.cogroup.Joiner class.

All input data comes from, and all output data feeds to, a cascading.tap.Tap instance.

A Tap represents a resource like a data file on the local file system, on a Hadoop distributed file system, or even on Amazon S3. Taps can be read from, which makes it a "source", or written to, which makes it a "sink". Or, more commonly, Taps can act as both sinks and sources when shared between Flows.

All Taps must have a Scheme associated with them. If the Tap is about where the data is, and how to get it, the Scheme is about what the data is. Cascading provides two Scheme classes, TextLine and SequenceFile.

The fundamental difference behind TextLine and SequenceFile schemes is that tuples stored in the SequenceFile remain tuples, so when read, they do not need to be parsed. So a typical Cascading application will read text files, and parse each line into a Tuple for processing. The final Tuples are saved via the SequenceFile scheme so future applications can just read the file directly into Tuple instances without the parsing step.

Tap tap = new Hfs( new TextLine( new Fields( "line" ) ), path );

The above example creates a new Hadoop FileSystem Tap that can read/write text files. Since only one field name was provided, the "offset" field is discarded, resulting in an input tuple stream with only "line" values.

The three most common Tap classes used are, Hfs, Dfs, and Lfs. The MultiTap and TemplateTap are utility Taps.

Keep in mind Hadoop cannot source data from directories with nested sub-directories, and it cannot write to directories that already exist. But you can simply point the Hfs Tap to a directory of data files and they all will be used as input, no need to enumate each individual file into a MultiTap.

To get around the last issue, the Hadoop related Taps allow for a SinkMode value to be set when constructed.

Tap tap = new Hfs( new TextLine( new Fields( "line" ) ), path, SinkMode.REPLACE );

Here are all the modes available by the built-in Tap types.

When pipe assemblies are bound to source and sink Taps, a Flow is created. Flows are executable in the sense that once created they can be "started" and will begin execution on a configured Hadoop cluster.

Think of a Flow as a data processing workflow that reads data from sources, processes the data as defined by the pipe assembly, and writes data to the sinks. Input source data does not need to exist when the Flow is created, but it must exist when the Flow is executed (unless executed as part of a Cascade, see Cascades).

The most common pattern is to create a Flow from an existing pipe assembly. But there are cases where a MapReduce job has already been created and it makes sense to encapsulate it in a Flow class so that it may participate in a Cascade and be scheduled with other Flow instances. Both patterns are covered here.

Flow flow = new FlowConnector().connect( "flow-name", source, sink, pipe );

// the "left hand side" assembly head
Pipe lhs = new Pipe( "lhs" );

lhs = new Each( lhs, new SomeFunction() );
lhs = new Each( lhs, new SomeFilter() );

// the "right hand side" assembly head
Pipe rhs = new Pipe( "rhs" );

rhs = new Each( rhs, new SomeFunction() );

// joins the lhs and rhs
Pipe join = new CoGroup( lhs, rhs );

join = new Every( join, new SomeAggregator() );

join = new GroupBy( join );

join = new Every( join, new SomeAggregator() );

// the tail of the assembly
join = new Each( join, new SomeFunction() );

Tap lhsSource = new Hfs( new TextLine(), "lhs.txt" );
Tap rhsSource = new Hfs( new TextLine(), "rhs.txt" );

Tap sink = new Hfs( new TextLine(), "output" );

Map<String, Tap> sources = new HashMap<String, Tap>();

sources.put( "lhs", lhsSource );
sources.put( "rhs", rhsSource );

Flow flow = new FlowConnector().connect( "flow-name", sources, sink, join );

Properties properties = new Properties();

// pass in the class name of your application
// this will find the parent jar at runtime
FlowConnector.setApplicationJarClass( properties, Main.class );

// or pass in the path to the parent jar
FlowConnector.setApplicationJarPath( properties, pathToJar );

FlowConnector flowConnector = new FlowConnector( properties );

If a MapReduce job already exists and needs to be managed by a Cascade, then the cascading.flow.MapReduceFlow class should be used. After creating a Hadoop JobConf instance, just pass it into the MapReduceFlow constructor. The resulting Flow instance can be used like any other Flow.

A Cascade allows multiple Flow instances to be executed as a single logical unit. If there are dependencies between the Flows, they will be executed in the correct order. Further, Cascades act like ant build or Unix "make" files. When run, a Cascade will only execute Flows that have stale sinks (output data that is older than the input data), by default.

CascadeConnector connector = new CascadeConnector();
Cascade cascade = connector.connect( flowFirst, flowSecond, flowThird );

When passing Flows to the CascadeConnector, order is not important. The CascadeConnector will automatically determine what the dependencies are between the given Flows and create a scheduler that will start each flow as its data sources become available. If two or more Flow instances have no dependencies, they will be submitted together so they can execute in parallel.

For more information, see the section on Topological Scheduling.

If an instance of cascading.flow.FlowSkipStrategy is given to an Cascade instance via the Cascade#setFlowSkipStrategy() method, it will be consulted for every Flow instance managed by the Cascade, all skip strategies on the Flow instances will be ignored. For more information on skip strategies, see Skipping Flows.

Cascading requires Hadoop to be installed and correctly configured. Apache Hadoop is an Open Source Apache project and is freely available. It can be downloaded from the Hadoop website, http://hadoop.apache.org/core/.

Cascading ships with a handful of jars.

Cascading will run with Hadoop in its default 'local' or 'stand alone' mode, or configured as a distributed cluster.

When used on a cluster, a Hadoop job Jar must be created with Cascading jars and dependent thrid-party jars in the job jar lib directory, per the Hadoop documentation.

<!-- Common ant build properties, included here for completeness -->
<property name="build.dir" location="${basedir}/build"/>
<property name="build.classes" location="${build.dir}/classes"/>

<!-- Cascading specific properties -->
<property name="cascading.home" location="${basedir}/../cascading"/>
<property file="${cascading.home}/version.properties"/>
<property name="cascading.release.version" value="x.y.z"/>
<property name="cascading.filename.core"
          value="cascading-core-${cascading.release.version}.jar"/>
<property name="cascading.filename.xml"
          value="cascading-xml-${cascading.release.version}.jar"/>
<property name="cascading.libs" value="${cascading.home}/lib"/>
<property name="cascading.libs.core" value="${cascading.libs}"/>
<property name="cascading.libs.xml" value="${cascading.libs}/xml"/>

<condition property="cascading.path" value="${cascading.home}/"
           else="${cascading.home}/build">
  <available file="${cascading.home}/${cascading.filename.core}"/>
</condition>

<property name="cascading.lib.core"
          value="${cascading.path}/${cascading.filename.core}"/>
<property name="cascading.lib.xml"
          value="${cascading.path}/${cascading.filename.xml}"/>

<!--
  A sample target to jar project classes and Cascading
  libraries into a single Hadoop compatible jar file.
 -->

<target name="jar" description="creates a Hadoop ready jar w/dependencies">

  <!-- copy Cascading classes and libraries -->
  <copy todir="${build.classes}/lib" file="${cascading.lib.core}"/>
  <copy todir="${build.classes}/lib" file="${cascading.lib.xml}"/>
  <copy todir="${build.classes}/lib">
    <fileset dir="${cascading.libs.core}" includes="*.jar"/>
    <fileset dir="${cascading.libs.xml}" includes="*.jar"/>
  </copy>

  <jar jarfile="${build.dir}/${ant.project.name}.jar">
    <fileset dir="${build.classes}"/>
    <fileset dir="${basedir}" includes="lib/"/>
    <manifest>
      <!-- the project Main class, by default assumes Main -->
      <attribute name="Main-Class" value="${ant.project.name}/Main"/>
    </manifest>
  </jar>

</target>

The above Ant snippets can be used in your project to create a Hadoop jar for submission on your cluster. Again, all Hadoop applications that are intended to be run in a cluster must be packaged with all third-party libraries in a directory named lib in the final application Jar file, regardless if they are Cascading applications or raw Hadoop MapReduce applications.

Note, the snippets above is only intended to show how to include Cascading libraries, you still need to compile your project into the build.classes path.

During runtime, Hadoop must be "told" which application jar file should be pushed to the cluster. Typically this is done via the Hadoop API JobConf object.

Cascading offers a shorthand for configuring this parameter.

Properties properties = new Properties();

// pass in the class name of your application
// this will find the parent jar at runtime
FlowConnector.setApplicationJarClass( properties, Main.class );

// or pass in the path to the parent jar
FlowConnector.setApplicationJarPath( properties, pathToJar );

FlowConnector flowConnector = new FlowConnector( properties );

Above we see how to set the same property two ways. First via the setApplicationJarClass() method, and via the setApplicationJarPath() method. The first method takes a Class object that owns the 'main' function for this application. The assumption here is that Main.class is not located in a Java Jar that is stored in the lib folder of the application Jar. If it is, that Jar will be pushed to the cluster, not the parent application jar.

In your application, only one of these methods needs to be called, but one of them must be called to properly configure Hadoop.

Running a Cascading application is exactly the same as running any Hadoop application. After packaging your application into a single jar (see Building Cascading Applications), you must use bin/hadoop to submit the application to the cluster.

For example, to execute an application stuffed into your-application.jar, call the Hadoop shell script:

$HADOOP_HOME/bin/hadoop jar your-application.jar [some params]

If the configuration scripts in $HADOOP_CONF_DIR are configured to use a cluster, the Jar will be pushed into the cluster for execution.

Cascading does not rely on any environment variables like $HADOOP_HOME or $HADOOP_CONF_DIR, only bin/hadoop does.

It should be noted that even though your-application.jar is passed on the command line to bin/hadoop this in no way configures Hadoop to push this jar into the cluster. You must still call one of the property setters mentioned above to set the proper path to the application jar. If misconfigured, likely one of the internal libraries (found in the lib folder) will be pushed to the cluster instead and ClassNotFoundExceptions will be thrown.

To use Cascading, it is not strictly necessary to create custom Operations. There are a number of Operations in the Cascading library that can be combined into very robust applications. In the same way you can chain sed, grep, sort, uniq, awk, etc in Unix, you can chain existing Cascading operations. But developing customs Operations is very simple in Cascading.

There are four kinds of Operations: Function, Filter, Aggregator, and Buffer.

All Operations operate on an input argument Tuple and all Operations other than Filter may return zero or more Tuple object results. That is, a Function can parse a string and return a new Tuple for every value parsed out (one Tuple for each 'word'), or it may create a single Tuple with every parsed value as an element in the Tuple object (one Tuple with "first-name" and "last-name" fields).

In practice, a Function that returns no results is a Filter, but the Filter type has been optimized and can be combined with "logical" filter Operations like Not, And, Or, etc.

During runtime, Operations actually receive arguments as an instance of the TupleEntry object. The TupleEntry object holds both an instance of Fields and the current Tuple the Fields object defines fields for. TupleEntry is a helper class that allows for tuple operations by using simple field names and is typically only exposed to developers writing custom Operations.

All Operations, other than Filter, must declare result Fields. For example, if a Function was written to parse words out of a String and return a new Tuple for each word, this Function must declare that it intends to return a Tuple with one field named "word". If the Function mistakenly returns more values in the Tuple other than a 'word', the process will fail. Operations that do return arbitrary numbers of values in a result Tuple may declare Fields.UNKNOWN.

The Cascading planner always attempts to "fail fast" where possible by checking the field name dependencies between Pipes and Operations, but some cases the planner can't account for.

All Operations must be wrapped by either an Each or an Every pipe instance. The pipe is responsible for passing in an argument Tuple and accepting the result Tuple.

A Function expects a single argument Tuple, and may return zero or more result Tuples.

A Function may only be used with a Each pipe which may follow any other pipe type.

To create a custom Function, subclass the class cascading.operation.BaseOperation and implement the interface cascading.operation.Function. Because BaseOperation has been subclassed, the operate method, as defined on the Function interface, is the only method that must be implemented.

public class SomeFunction extends BaseOperation implements Function
  {
  public void operate( FlowProcess flowProcess, FunctionCall functionCall )
    {
    // get the arguments TupleEntry
    TupleEntry arguments = functionCall.getArguments();

    // create a Tuple to hold our result values
    Tuple result = new Tuple();

    // insert some values into the result Tuple

    // return the result Tuple
    functionCall.getOutputCollector().add( result );
    }
  }

Functions should declare both the number of argument values they expect, and the field names of the Tuple they will return.

Functions must accept 1 or more values in a Tuple as arguments, by default they will accept any number (Operation.ANY) of values. Cascading will verify that the number of arguments selected match the number of arguments expected during the planning phase.

Functions may optionally declare the field names they return, by default Functions declare Fields.UNKNOWN.

Both declarations must be done on the constructor, either by passing default values to the super constructor, or by accepting the values from the user via a constructor implementation.

public class AddValuesFunction extends BaseOperation implements Function
  {
  public AddValuesFunction()
    {
    // expects 2 arguments, fail otherwise
    super( 2, new Fields( "sum" ) );
    }

  public AddValuesFunction( Fields fieldDeclaration )
    {
    // expects 2 arguments, fail otherwise
    super( 2, fieldDeclaration );
    }

  public void operate( FlowProcess flowProcess, FunctionCall functionCall )
    {
    // get the arguments TupleEntry
    TupleEntry arguments = functionCall.getArguments();

    // create a Tuple to hold our result values
    Tuple result = new Tuple();

    // sum the two arguments
    int sum = arguments.getInteger( 0 ) + arguments.getInteger( 1 );

    // add the sum value to the result Tuple
    result.add( sum );

    // return the result Tuple
    functionCall.getOutputCollector().add( result );
    }
  }

The example above implements a fully functional Function that accepts two values in the argument Tuple, adds them together, and returns the result in a new Tuple.

The first constructor assumes a default field name this function will return, but it is a best practice to always give the user the option to override the declared field names to prevent any field name collisions that would cause the planner to fail.

A Filter expects a single argument Tuple and returns a boolean value stating whether or not the current Tuple in the tuple stream should be discarded.

A Filter may only be used with a Each pipe, and it may follow any other pipe type.

To create a custom Filter, subclass the class cascading.operation.BaseOperation and implement the interfacecascading.operation.Filter. Because BaseOperation has been subclassed, the isRemove method, as defined on the Filter interface, is the only method that must be implemented.

public class SomeFilter extends BaseOperation implements Filter
  {
  public boolean isRemove( FlowProcess flowProcess, FilterCall filterCall )
    {
    // get the arguments TupleEntry
    TupleEntry arguments = filterCall.getArguments();

    // initialize the return result
    boolean isRemove = false;

    // test the argument values and set isRemove accordingly

    return isRemove;
    }
  }

Filters should declare the number of argument values they expect.

Filters must accept 1 or more values in a Tuple as arguments, by default they will accept any number ( Operation.ANY) of values. Cascading will verify the number of arguments selected match the number of arguments expected.

The number of arguments declarations must be done on the constructor, either by passing a default value to the super constructor, or by accepting the value from the user via a constructor implementation.

public class StringLengthFilter extends BaseOperation implements Filter
  {
  public StringLengthFilter()
    {
    // expects 2 arguments, fail otherwise
    super( 2 );
    }

  public boolean isRemove( FlowProcess flowProcess, FilterCall filterCall )
    {
    // get the arguments TupleEntry
    TupleEntry arguments = filterCall.getArguments();

    // filter out the current Tuple if the first argument length is greater
    // than the second argument integer value
    return arguments.getString( 0 ).length() > arguments.getInteger( 1 );
    }
  }

The example above implements a fully functional Filter that accepts two arguments and filters out the current Tuple if the first argument String length is greater than the integer value of the second argument.

An Aggregator expects set of argument Tuples in the same grouping, and may return zero or more result Tuples.

An Aggregator may only be used with an Every pipe, and it may only follow a GroupBy,CoGroup, or another Every pipe type.

To create a custom Aggregator, subclass the class cascading.operation.BaseOperation and implement the interfacecascading.operation.Aggregator. Because BaseOperation has been subclassed, the start, aggregate, and complete methods, as defined on the Aggregator interface, are the only methods that must be implemented.

public class SomeAggregator extends BaseOperation<SomeAggregator.Context>
  implements Aggregator<SomeAggregator.Context>
  {
  public static class Context
    {
    Object value;
    }

  public void start( FlowProcess flowProcess,
                     AggregatorCall<Context> aggregatorCall )
    {
    // get the group values for the current grouping
    TupleEntry group = aggregatorCall.getGroup();

    // create a new custom context object
    Context context = new Context();

    // optionally, populate the context object

    // set the context object
    aggregatorCall.setContext( context );
    }

  public void aggregate( FlowProcess flowProcess,
                         AggregatorCall<Context> aggregatorCall )
    {
    // get the current argument values
    TupleEntry arguments = aggregatorCall.getArguments();

    // get the context for this grouping
    Context context = aggregatorCall.getContext();

    // update the context object
    }

  public void complete( FlowProcess flowProcess,
                        AggregatorCall<Context> aggregatorCall )
    {
    Context context = aggregatorCall.getContext();

    // create a Tuple to hold our result values
    Tuple result = new Tuple();

    // insert some values into the result Tuple based on the context

    // return the result Tuple
    aggregatorCall.getOutputCollector().add( result );
    }
  }

Aggregators should declare both the number of argument values they expect, and the field names of the Tuple they will return.

Aggregators must accept 1 or more values in a Tuple as arguments, by default they will accept any number ( Operation.ANY) of values. Cascading will verify the number of arguments selected match the number of arguments expected.

Aggregators may optionally declare the field names they return, by default Aggregators declare Fields.UNKNOWN.

Both declarations must be done on the constructor, either by passing default values to the super constructor, or by accepting the values from the user via a constructor implementation.

public class AddTuplesAggregator
    extends BaseOperation<AddTuplesAggregator.Context>
    implements Aggregator<AddTuplesAggregator.Context>
  {
  public static class Context
    {
    long value = 0;
    }

  public AddTuplesAggregator()
    {
    // expects 1 argument, fail otherwise
    super( 1, new Fields( "sum" ) );
    }

  public AddTuplesAggregator( Fields fieldDeclaration )
    {
    // expects 1 argument, fail otherwise
    super( 1, fieldDeclaration );
    }

  public void start( FlowProcess flowProcess,
                     AggregatorCall<Context> aggregatorCall )
    {
    // set the context object, starting at zero
    aggregatorCall.setContext( new Context() );
    }

  public void aggregate( FlowProcess flowProcess,
                         AggregatorCall<Context> aggregatorCall )
    {
    TupleEntry arguments = aggregatorCall.getArguments();
    Context context = aggregatorCall.getContext();

    // add the current argument value to the current sum
    context.value += arguments.getInteger( 0 );
    }

  public void complete( FlowProcess flowProcess,
                        AggregatorCall<Context> aggregatorCall )
    {
    Context context = aggregatorCall.getContext();

    // create a Tuple to hold our result values
    Tuple result = new Tuple();

    // set the sum
    result.add( context.value );

    // return the result Tuple
    aggregatorCall.getOutputCollector().add( result );
    }
  }

The example above implements a fully functional Aggregator that accepts one value in the argument Tuple, adds all these argument Tuples in the current grouping, and returns the result as a new Tuple.

The first constructor assumes a default field name this Aggregator will return, but it is a best practice to always give the user the option to override the declared field names to prevent any field name collisions that would cause the planner to fail.

A Buffer expects set of argument Tuples in the same grouping, and may return zero or more result Tuples.

The Buffer is very similar to an Aggregator except it receives the current Grouping Tuple and an iterator of all the arguments it expects for every value Tuple in the current grouping, all on the same method call. This is very similar to the typical Reducer interface, and is best used for operations that need greater visibility to the previous and next elements in the stream. For example, smoothing a series of time-stamps where there are missing values.

An Buffer may only be used with an Every pipe, and it may only follow a GroupBy or CoGroup pipe type.

To create a custom Buffer, subclass the class cascading.operation.BaseOperation and implement the interfacecascading.operation.Buffer. Because BaseOperation has been subclassed, the operate method, as defined on the Buffer interface, is the only method that must be implemented.

public class SomeBuffer extends BaseOperation implements Buffer
  {
  public void operate( FlowProcess flowProcess, BufferCall bufferCall )
    {
    // get the group values for the current grouping
    TupleEntry group = bufferCall.getGroup();

    // get all the current argument values for this grouping
    Iterator<TupleEntry> arguments = bufferCall.getArgumentsIterator();

    // create a Tuple to hold our result values
    Tuple result = new Tuple();

    while( arguments.hasNext() )
      {
      TupleEntry argument = arguments.next();

      // insert some values into the result Tuple based on the arguemnts
      }

    // return the result Tuple
    bufferCall.getOutputCollector().add( result );
    }
  }

Buffer should declare both the number of argument values they expect, and the field names of the Tuple they will return.

Buffers must accept 1 or more values in a Tuple as arguments, by default they will accept any number ( Operation.ANY) of values. Cascading will verify the number of arguments selected match the number of arguments expected.

Buffers may optionally declare the field names they return, by default Buffers declare Fields.UNKNOWN.

Both declarations must be done on the constructor, either by passing default values to the super constructor, or by accepting the values from the user via a constructor implementation.

public class AverageBuffer extends BaseOperation implements Buffer
  {

  public AverageBuffer()
    {
    super( 1, new Fields( "average" ) );
    }

  public AverageBuffer( Fields fieldDeclaration )
    {
    super( 1, fieldDeclaration );
    }

  public void operate( FlowProcess flowProcess, BufferCall bufferCall )
    {
    // init the count and sum
    long count = 0;
    long sum = 0;

    // get all the current argument values for this grouping
    Iterator<TupleEntry> arguments = bufferCall.getArgumentsIterator();

    while( arguments.hasNext() )
      {
      count++;
      sum += arguments.next().getInteger( 0 );
      }

    // create a Tuple to hold our result values
    Tuple result = new Tuple( sum / count );

    // return the result Tuple
    bufferCall.getOutputCollector().add( result );
    }
  }

The example above implements a fully functional buffer that accepts one value in argument Tuple, adds all these argument Tuples in the current grouping, and returns the result divided by the number of argument tuples counted in a new Tuple.

The first constructor assumes a default field name this Buffer will return, but it is a best practice to always give the user the option to override the declared field names to prevent any field name collisions that would cause the planner to fail.

Note this example is somewhat fabricated, in practice a Aggregator should be implemented to compute averages. A Buffer would be better suited for "running averages" across very large spans, for example.

In all the above sections, the cascading.operation.BaseOperation class was subclassed. This class is an implementation of the cascading.operation.Operation interface and provides a few default method implementations. It is not strictly required to extend BaseOperation, but it is very convenient to do so.

When developing custom operations, the developer may need to initialize and destroy a resource. For example, when doing pattern matching, a java.util.regex.Matcher may need to be initialized and used in a thread-safe way. Or a remote connection may need to be opened and eventually closed. But for performance reasons, the operation should not create/destroy the connection for each Tuple or every Tuple group that passes through.

The interface Operation declares two methods, prepare() and cleanup(). In the case of Hadoop and MapReduce, the prepare() and cleanup() methods are called once per Map or Reduce task. prepare() is called before any argument Tuple is passed in, and cleanup() is called after all Tuple arguments have been operated on. Within each of these methods, the developer can initialize a "context" object that can hold an open socket connection, or Matcher instance. The "context" is user defined and is the same mechanism used by the Aggregator operation, except the Aggregator is also given the opportunity to initialize and destroy its context via the start() and complete() methods.

If a "context" object is used, its type should be declared in the sub-class class declaration using the Java Generics notation.

Cascading SubAssemblies are reusable pipe assemblies that are linked into larger pipe assemblies. Think of them as subroutines in a programming language. The help organize complex pipe assemblies and allow for commonly used pipe assemblies to be packaged into libraries for inclusion by other users.

To create a SubAssembly, the cascading.pipe.SubAssembly class must be subclassed.

public class SomeSubAssembly extends SubAssembly
  {
  public SomeSubAssembly( Pipe lhs, Pipe rhs )
    {
    // continue assembling against lhs
    lhs = new Each( lhs, new SomeFunction() );
    lhs = new Each( lhs, new SomeFilter() );

    // continue assembling against lhs
    rhs = new Each( rhs, new SomeFunction() );

    // joins the lhs and rhs
    Pipe join = new CoGroup( lhs, rhs );

    join = new Every( join, new SomeAggregator() );

    join = new GroupBy( join );

    join = new Every( join, new SomeAggregator() );

    // the tail of the assembly
    join = new Each( join, new SomeFunction() );

    // must register all assembly tails
    setTails( join );
    }
  }

In the above example, we pass in via the constructor pipes we wish to continue assembling against, and the last line we register the 'join' pipe as a tail. This allows SubAssemblies to be nested within larger pipe assemblies or other SubAssemblies.

// the "left hand side" assembly head
Pipe lhs = new Pipe( "lhs" );

// the "right hand side" assembly head
Pipe rhs = new Pipe( "rhs" );

// our custom SubAssembly
Pipe pipe = new SomeSubAssembly( lhs, rhs );

pipe = new Each( pipe, new SomeFunction() );

Above we see how natural it is to include a SubAssembly into a new pipe assembly.

If we had a SubAssembly that represented a split, that is, had two or more tails, we could use the getTails() method to get at the array of "tails" set internally by the setTails() method.

public class SplitSubAssembly extends SubAssembly
  {
  public SplitSubAssembly( Pipe pipe )
    {
    // continue assembling against lhs
    pipe = new Each( pipe, new SomeFunction() );

    Pipe lhs = new Pipe( "lhs", pipe );
    lhs = new Each( lhs, new SomeFunction() );

    Pipe rhs = new Pipe( "rhs", pipe );
    rhs = new Each( rhs, new SomeFunction() );

    // must register all assembly tails
    setTails( lhs, rhs );
    }
  }

// the "left hand side" assembly head
Pipe head = new Pipe( "head" );

// our custom SubAssembly
SubAssembly pipe = new SplitSubAssembly( head );

// grab the split branches
Pipe lhs = new Each( pipe.getTails()[0], new SomeFunction() );
Pipe rhs = new Each( pipe.getTails()[1], new SomeFunction() );

To rephrase, if a SubAssembly does not split the incoming Tuple stream, the SubAssembly instance can be passed directly to the next Pipe instance. But, if the SubAssembly splits the stream into multiple branches, each branch tail must be passed to the setTails() method, and the getTails() method should be called to get a handle to the correct branch to pass to the next Pipe instances.

Stream assertions are simply a mechanism to 'assert' that one or more values in a tuple stream meet certain criteria. This is similar to the Java language 'assert' keyword, or a unit test. An example would be 'assert not null' or 'assert matches'.

Assertions are treated like any other function or aggregator in Cascading. They are embedded directly into the pipe assembly by the developer. If an assertion fails, the processing stops, by default. Alternately they can trigger a Failure Trap.

As with any test, sometimes they are wanted, and sometimes they are unnecessary. Thus stream assertions are embedded as either 'strict' or 'validating'.

When running a tests against regression data, it makes sense to use strict assertions. This regression data should be small and represent many of the edge cases the processing assembly must support robustly. When running tests in staging, or with data that may vary in quality since it is from an unmanaged source, using validating assertions make much sense. Then there are obvious cases where assertions just get in the way and slow down processing and it would be nice to just bypass them.

During runtime, Cascading can be instructed to plan out strict, validating, or all assertions before building the final MapReduce jobs via the MapReduce Job Planner. And they are truly planned out of the resulting job, not just switched off, providing the best performance.

This is just one feature of lazily building MapReduce jobs via a planner, instead of hard coding them.

// incoming -> "ip", "time", "method", "event", "status", "size"

AssertNotNull notNull = new AssertNotNull();
assembly = new Each( assembly, AssertionLevel.STRICT, notNull );

AssertSizeEquals equals = new AssertSizeEquals( 6 );
assembly = new Each( assembly, AssertionLevel.STRICT, equals );

AssertMatchesAll matchesAll = new AssertMatchesAll( "(GET|HEAD|POST)" );
assembly = new Each( assembly, new Fields("method"),
                     AssertionLevel.STRICT, matchesAll );

// outgoing -> "ip", "time", "method", "event", "status", "size"

Again, assertions are added to a pipe assembly like any other operation, except the AssertionLevel must be set, so the planner knows how to treat the assertion during planning.

Properties properties = new Properties();

// removes all assertions from the Flow
FlowConnector.setAssertionLevel( properties, AssertionLevel.NONE );

FlowConnector flowConnector = new FlowConnector( properties );

Flow flow = flowConnector.connect( source, sink, assembly );

To configure the planner to remove some or all assertions, a property must be set via the FlowConnector#setAssertionLevel() method. AssertionLevel.NONE removes all assertions. AssertionLevel.VALID keeps VALID assertions but removes STRICT ones. And AssertionLevel.STRICT keeps all assertions, which is the planner default value.

Failure Traps are the same as a Tap sink (opposed to a source), except being bound to a particular tail element of the pipe assembly, traps can be bound to intermediate pipe assembly segments, like to a Stream Assertion.

Whenever an operation fails and throws an exception, and there is an associated trap, the offending Tuple will be saved to the resource specified by the trap Tap. This allows the job to continue processing without any data loss.

By design, clusters are hardware fault tolerant. Lose a node, the cluster continues working.

But software fault tolerance is a little different. Failure Traps provide a means for the processing to continue without losing track of the data that caused the fault. For high fidelity applications, this may not be so attractive, but low fidelity applications (like web page indexing) this can dramatically improve processing reliability.

// ...some useful pipes here

// name this pipe assembly segment
assembly = new Pipe( "assertions", assembly );

AssertNotNull notNull = new AssertNotNull();
assembly = new Each( assembly, AssertionLevel.STRICT, notNull );

AssertSizeEquals equals = new AssertSizeEquals( 6 );
assembly = new Each( assembly, AssertionLevel.STRICT, equals );

AssertMatchesAll matchesAll = new AssertMatchesAll( "(GET|HEAD|POST)" );
assembly = new Each( assembly, new Fields("method"),
                     AssertionLevel.STRICT, matchesAll );

// ...some more useful pipes here

Map<String,Tap> traps = new HashMap<String,Tap>();

traps.put( "assertions", trap );

FlowConnector flowConnector = new FlowConnector();
Flow flow =
  flowConnector.connect( "log-parser", source, sink, traps, assembly );

In the above example, we bind our trap Tap to the pipe assembly segment named "assertions". Note how we can name branches and segments by using a single Pipe instance and it applies to all subsequent Pipe instances.

Each Flow, has the ability to execute callbacks via an event listener. This is very useful when external application need to be notified that a Flow has completed.

A good example is when running Flows on an Amazon EC2 Hadoop cluster. After the Flow is completed, a SQS event can be sent notifying another application it can now fetch the job results from S3. In tandem, it can start the process of shutting down the cluster if no more work is queued up for it.

Flows support event listeners through the cascading.flow.FlowListener interface. The FlowListener interface supports four events, onStarting, onStopping, onCompleted, and onThrowable.

FlowListeners are useful when external systems must be notified when a Flow has completed or failed.

The TemplateTap Tap class provides a simple means to break large datasets into smaller sets based on values in the dataset. Typically this is called 'binning' the data, where each 'bin' of data is named after values shared by the data in that bin. For example, organizing log files by month and year.

Hfs tap = new Hfs( new TextLine( new Fields( "year", "month", "entry" ) ), path );

String template = "%s-%s"; // dirs named "year-month"
Tap months = new TemplateTap( tap, template, SinkMode.REPLACE );

In the above example, we construct a parent Hfs Tap and pass it to the constructor of a TemplateTap instance along with a String format 'template'. This format template is populated in the order values are declared via the Scheme class. If more complex path formatting is necessary then you may subclass the TemplateTap.

Note that you can only create sub-directories to bin data into. Hadoop must still write 'part' files into each bin directory.

One last thing to keep in mind is whether or not 'binning' happens during the Map or Reduce phase. By doing a GroupBy on the values that will be used to populate the template, binning will happen during the Reduce phase and likely scale much better if there are a very large number of unique grouping keys.

Cascading was designed with scripting in mind. Since it is just an API, any Java compatible scripting language can import and instantiate Cascading classes and create pipe assemblies, flows, and execute those flows.

And if the scripting language in question supports Domain Specific Language (DSL) creation, the user can create her own DSL to handle common idioms.

See the Cascading website for publicly available scripting language bindings.

Cascading was designed to be easily configured and enhanced by developers. Besides allowing for custom Operations, developers can provide custom Tap and Scheme types so applications can connect to system external to Hadoop.

A Tap represents something "physical", like a file or a database table. Subsequently Tap implementations are responsible for life cycle issues around the resource they represent, like tests for existence, or deleting.

A Scheme represents a format or representation, like a text format for a file, or columns in a table. Schemes are responsible for converting the Tap managed resources proprietary format to and from a cascading.tuple.Tuple instance.

Unfortunately creating custom Taps and Schemes can be an involved process and requires some knowledge of Hadoop and the Hadoop FileSystem API. Most commonly, the cascading.tap.Hfs class can be subclassed if a new file system is to be supported, assuming passing a fully qualified URL to the Hfs constructor isn't sufficient (the Hfs tap will look up a file system based on the URL scheme via the Hadoop FileSystem API).

Delegating to the Hadoop FileSystem API is not a strict requirement, but the developer will need to implement a Hadoop org.apache.hadoop.mapred.InputFormat and/or .org.apache.hadoop.mapred.OutputFormat so that Hadoop knows how to split and handle the incoming/outgoing data. The custom Scheme is responsible for setting InputFormat and OutputFormat on the JobConf via the sinkInit and sourceInit methods.

For examples on how to implement a custom Tap and Scheme, see the Cascading Modules page for samples.

The cascading.operation.Identify function is used to "shape" a tuple stream. Here are some common patterns.

The cascading.operation.Debug function is a utility Function (actually, its a Filter) that will print the current argument Tuple to either stdout or stderr.

It is only intended to be used during development to help debug a difficult pipe assembly.

The Sample and Limit functions are used to limit the number of Tuples that pass through a pipe assembly.

The cascading.operation.Insert function allows for insertion of literal values into the tuple stream.

This is most useful when a splitting a tuple stream and one of the branches needs some identifying value. Or when some missing parameter or value, like a date String for todays date, needs to be inserted.

Cascading includes a number of text functions in the cascading.operation.text package.

Cascading provides some support for dynamically compiled Java expression to be used as either Functions or Filters. This functionality is provided by the Janino embedded compiler. Janino and its documentation can be found on its website, http://www.janino.net/. But in short, an Expression is a single line of Java, for example a + 3 * 2 or a < 7. The first would resolve to some number, the second to a boolean value. Where a and b are field names passed in as Tuple arguments to the Operation. Janino will compile this expression into byte code giving compiled code processing speeds.

All XML Operations are kept in a module other than core, so can be included in a Cascading application by including the cascading-xml-x.y.z.jar in the project. This module has one dependency, the TagSoup library, which allows for HTML and XML "tidying". More about TagSoup can be read on its website, http://home.ccil.org/~cowan/XML/tagsoup/.

Cascading Stream Assertions are used to build robust reusable pipe assemblies. They can be planned out of a Flow instance during runtime. For more information see the section on Stream Assertions. Below we describe the Assertions available in the core library.

The logical Filter operators allows the user to assemble more complex filtering to be used in a single Pipe, instead of chaining multiple Pipes together to get the same effect.

// incoming -> "ip", "time", "method", "event", "status", "size"

FilterNull filterNull = new FilterNull();
RegexFilter regexFilter = new RegexFilter( "(GET|HEAD|POST)" );

And andFilter = new And( filterNull, regexFilter );

assembly = new Each( assembly, new Fields( "method" ), andFilter );

// outgoing -> "ip", "time", "method", "event", "status", "size"

Above, we are "and-ing" the two filters. Both must be satisfied for the data to pass through this one Pipe.

Testing Operations, pipe-assemblies, and applications is a must. The cascading.CascadingTestCase provides a number of helper methods.

When testing custom Operations, use the invokeFunction(), invokeFilter(), invokeAggregator(), and invokeBuffer() methods.

When testing Flows, use the validateLength() methods. There are quite a few, each offering extra flexibility. All of them will read the sink Tap and validate it is the correct length, have the correct Tuple size, and if the values match a given regular expression pattern.

The cascading.ClusterTestCase can be used if you want to launch an embedded Hadoop cluster inside your TestCase.

Make sure cascading-test-x.y.z.jar is in your testing class-path in order to use these helper classes.

When developing your applications, use SubAssembly sub-classes, not "factory" methods. This way the code is much easier to read and to test.

The funny thing is that Object constructors are "factories", so there isn't much reason to build frameworks to duplicate what a constructor already does. Of course there are exceptions, but in practice they are rare when you can use a SubAssembly.

SubAssembies provide a very convenient means to co-locate like responsibilities into a single place. For example, have a ParsingSubAssembly and a RulesSubAssembly, where the first is responsible solely for parsing incoming Tuple streams (log files for example), and the second applied rules to decide if a given Tuple should be discarded or marked as bad.

Further, in your unit tests, you can create an TestAssertionsSubAssembly, that just inlines various ValueAssertions and GroupAssertions. Inlining Assertions directly in your SubAssemblies is also very important, but sometimes it makes sense to have more tests outside of the business logic.

There are a number of Operations in Cascading that will compile and apply Java expressions on the fly, see ExpressionFunction and ExpressionFilter for examples. In these expressions, Operation argument field names are used as variable in the expression. When creating field names, be conscious of the fact that if they are used in an expression, some characters will cause compilation errors. For example, "first-name" is a valid field name for use with Cascading, but this expression, first-name.trim(), will fail.

Oftentimes the FlowConnector will fail when attempting to plan a Flow. If the exception message given by PlannerException is vague, use the method PlannerException.writeDOT() to export a text representation of the internal pipe assembly. DOT files can be opened by GraphViz and OmnitGraffle. These plans are only partial, but you will be able to see where the Cascading planner failed.

Also note you can create a DOT file from a Flow as well via Flow.writeDOT().

The MapReduce Job Planner is an internal feature of Cascading.

When a collection of functions, splits, and joins are all tied up together into a 'pipe assembly', the FlowConnector object is used to create a new Flow instance against input and output data paths. This Flow is a single Cascading job.

Internally the FlowConnector employs an intelligent planner to convert the pipe assembly to a graph of dependent MapReduce jobs that can be executed on a Hadoop cluster.

All this happens under the scenes. As is the scheduling of the individual MapReduce jobs, and the clean up of intermediate data sets that bind the jobs together.

Above we can see how a reasonably normal Flow would be partitioned into MapReduce jobs. Every job is delimited by a temporary file that is the sink from the first job, and then the source to the next job.

To see how your Flows are partitioned, call the Flow#writeDOT() method. This will write a DOT file out to the path specified, and can be imported into a graphics package like OmniGraffle or Graphviz.

Cascading has a simple class, Cascade , that will take a collection of Cascading Flows and execute them on the target cluster in dependency order.

Consider the following example.

A Cascade is constructed through the CascadeConnector class, by building an internal graph that makes each Flow a 'vertex', and each file an 'edge'. A topological walk on this graph will touch each vertex in order of its dependencies. When a vertex has all it's incoming edges (files) available, it will be scheduled on the cluster.

In the example above, 'first' goes first, 'second' goes second, and 'third' is last.

If two or more Flows are independent of one another, they will be scheduled concurrently.

And by default, if any outputs from a Flow are newer than the inputs, the Flow is skipped. The assumption is that the Flow was executed recently, since the output isn't stale. So there is no reason to re-execute it and use up resources or add time to the job. This is similar behaviour a compiler would exhibit if a source file wasn't updated before a recompile.

This is very handy if you have a large number of jobs that should be executed as a logical unit with varying dependencies between them. Just pass them to the CascadeConnector, and let it sort them all out.