Cascading 3.0 User Guide - Advanced Processing

In Cascading, a SubAssembly is a reusable pipe assembly that can be joined with other instances of a SubAssembly to form a larger pipe assembly. SubAssemblies are much like subroutines in a larger program. SubAssemblies are a good way to organize complex pipe assemblies, and they allow for commonly used pipe assemblies to be packaged into libraries for inclusion in other projects by other users.

Many prebuilt SubAssemblies are available in the core Cascading library. See Built-in SubAssemblies for details.

To create a SubAssembly, subclass the cascading.pipe.SubAssembly class.

Notice that in Example 1:

Example 2 demonstrates how to include a SubAssembly in a new pipe assembly.

In a SubAssembly that represents a split — that is, a SubAssembly with two or more tails — you can use the getTails() method to access the array of tails set internally by the setTails() method.

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, handles will be needed to access them. The solution is to pass each branch tail to the setTails() method and to call the getTails() method to get handles for the desired branches. The handles can be passed to subsequent instances of Pipe.

Above we have inserted "assertion" pipes into the pipe assembly either between other pipes and/or taps.

Stream assertions are simply a mechanism for asserting 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. Command examples are assert not null and assert matches.

Assertions are treated like any other function or aggregator in Cascading. They are embedded directly into the pipe assembly by the developer. By default, if an assertion fails, the processing fails. As an alternative, an assertion failure can be caught by a failure Trap.

Assertions may be more, or less, desirable in different contexts. For this reason, stream assertions can be treated as either "strict" or "validating." Strict assertions make sense when running tests against regression data. These assemblies should be small and should represent many of the edge cases that the processing assembly must robustly support. Validating assertions, on the other hand, make more sense when running tests in staging or when using data that may vary in quality due to an unmanaged source.

And of course there are cases where assertions are unnecessary because they only would impede processing.

Cascading can be instructed to plan out (i.e., omit) strict assertions (leaving the validating assertions) or both strict and validating assertions when building the Flow. To create optimal performance, Cascading implements this by actually leaving the undesired assertions out of the final Flow (not merely disabling the assertions).

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

To configure the planner to remove some or all assertions, a property can be set via the FlowConnectorProps.setAssertionLevel() method or directly on the FlowDef instance. An example of setting the FlowDef instance is shown in Example 6.

Cascading provides the ability to trap the data and associated diagnostics that cause Java exceptions to be thrown from an Operation or Tap.

Typically if an exception is thrown cluster side, Cascading stops the complete executing Flow and forces the Flow.complete() method to throw an exception on the client side. Obviously if this exception is not handled, the client application will exit.

To prevent the shutdown, a trap Tap can be bound to whole branches. When an exception is encountered, the argument data is saved to the location specified by the trap Tap, including any specific diagnostic fields that may aid in resolving persistent issues.

The following diagram shows the use of traps in a pipe assembly.

Failure Traps are similar to tap sinks (as opposed to tap sources) in that they allow data to be stored. The difference is that Tap sinks are bound to a particular tail pipe in a pipe assembly and are the primary outlet of a branch in a pipe assembly. Traps can be bound to intermediate pipe assembly branches, but they only capture data that cause an Operation to fail (those that throw an exception).

Whenever an operation fails and throws an exception, if there is an associated trap, the offending Tuple is saved to the resource specified by the trap Tap. This allows the job to continue processing, while saving any "bad" data for future inspection.

By design, clusters are hardware fault-tolerant - lose a node, and the cluster continues working. But fault tolerance for software 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 very useful, since you likely will want any errors during processing to cause the application to stop. But for low-fidelity applications, such as webpage indexing, where skipping a page or two out of a few million is acceptable, this can dramatically improve processing reliability.

The example above binds a trap Tap to the pipe assembly segment named "assertions." Note how we can name branches and segments by using a single Pipe instance. The naming applies to all subsequent Pipe instances.

Traps are for exceptional cases, in the same way that Java Exception handling is. Traps are not intended for application flow control, and not a means to filter some data into other locations. Applications that need to filter out bad data should do so explicitly, using filters. For more on this, see Handling Good and Bad Data.

Optionally, the following diagnostic information may be captured along with the argument Tuple values.

See the cascading.tap.TrapProps Javadoc for more details.

Checkpointing is the ability to collapse a tuple stream within a Flow at any point as a way to improve the reliability or performance of a Flow. This is accomplished by using cascading.pipe.Checkpoint Pipe.

Checkpointing forces all tuple stream data to be written to disk, shared filesystem, or some other proprietary means provided by the underlying platform. The data is written at the end of a Pipe, prior to processing of the next Pipe in a stream.

By default a Checkpoint is anonymous and is cleaned up immediately after the Flow completes.

This feature is useful when used in conjunction with a HashJoin where the small side of the join starts out extremely large but is filtered down to fit into memory before being read into the HashJoin. By forcing a Checkpoint before the HashJoin, only the small filtered version of the data is replicated over the cluster. Without the Checkpoint, it is likely that the full, unfiltered file will be replicated to every node that the pipe assembly is executing.

On some platforms, checkpointing can allow for a Flow to be restarted after a transient failure. See Restarting a Checkpointed Flow below.

Alternatively, checkpointing is useful for debugging when used with a Checkpoint Tap, where the Tap has specified a TextDelimited Scheme without any declared Fields.

As can be seen above, we instantiate a new Checkpoint Tap by passing it the previous Every Pipe. This will be the point at which data is persisted.

In Example 8:

When using Checkpoint pipes in a Flow and the Flow fails, a future execution of the Flow can be restarted after the last successful FlowStep writing to a Checkpoint file. In other words, a Flow will only restart from the last Checkpoint Pipe location.

This feature requires that the following conditions are met:

Caution should be used when using restarted Checkpoint Flows. If the input data has changed or the pipe assembly has significantly been altered, the Flow may fail or there may be undetectable errors.

Note that when using a runID, all Flow instances must use a unique value except for those that attempt to restart the Flow. The runID value is used to scope the directories for the temporary checkpoint files to prevent file name collisions.

On successful completion of a Flow with a runID, any temporary checkpoint files are removed.

Each Flow and Cascade has the ability to execute callbacks via an event listener. This ability is useful when an external application needs to be notified that either a Flow or Cascade has started, halted, completed, or either has thrown an exception.

For instance, at the completion of a Flow that runs on an Amazon EC2 Hadoop cluster, an Amazon SQS message can be sent to notify another application to fetch the job results from S3 or begin the shutdown of the cluster.

Flows support event listeners through the cascading.flow.FlowListener interface. Cascades support event listeners through the cascading.cascade.CascadeListener, which supports four events:

The PartitionTap Tap class provides a simple means to break large data sets into smaller sets based on data item values. This is also commonly called binning the data, where each "bin" of data is named after some data value(s) shared by the members of that bin. For example, this is a simple way to organize log files by month and year.

In the example above, a parent FileTap tap is constructed and passed to the constructor of a PartitionTap instance, along with a cascading.tap.partition.DelimitedPartition "partitioner".

If more complex path formatting is necessary, you may implement the cascading.tap.partition.Partition interface.

It is important to see in the above example that the parentTap only sinks "entry" fields to a text-delimited file. But the monthsTap expects "year", "month", and "entry" fields from the tuple stream.

Here data is stored in the directory name for each partition when the PartitionTap is a sink, there is no need to redundantly store the data in the text delimited file. When reading from a PartitionTap, the directory name is parsed and its values are added to the outgoing tuple stream.

One last thing to keep in mind is where writing happens when executing on a cluster. By doing a GroupBy on the values used to define the partition, binning will happen during the grouping (reducer or partitioning) phase, and will likely scale much better in cases where there are a very large number of unique partitions that will result in a large number of directories or files.

Cascading implements a mechanism to perform partial aggregations in order to reduce the amount of transmitted data so that a complete aggregation can be completed down stream. This implementation allows any aggregate function to be implemented — not just Associative and Commutative functions.

Cascading provides a few built-in partial aggregate operations, including AverageBy, CountBy, SumBy, and FirstBy. These are actually SubAssemblies, not Operations, and are subclasses of the AggregateBy SubAssembly. For more on this, see the section on AggregateBy.

Using partial aggregate operations is quite easy. They are actually less verbose than a standard Aggregate operation.

For composing multiple partial aggregate operations, things are done a little differently.