Cascading 4.0 User Guide - The Apache Hadoop Platform

The Hadoop website describes the technology as "a software platform that lets one easily write and run applications that process vast amounts of data." Hadoop does this by providing a storage layer that holds vast amounts of data and an execution layer that runs an application in parallel across the cluster. The platform distributes the application load as subsets of the stored data across the cluster and coordinates the distributed processing tasks to optimize the compute resources of the environment.

The Hadoop Distributed File System (HDFS) is the storage layer, which serves as a single storage volume that is optimized for many concurrent serialized reads of large data files — where "large" might be measured in gigabytes or petabytes. However, HDFS does have limitations. For example, random access to the data is not really possible in an efficient manner. Also, Hadoop only supports a single writer for output. But this limit helps make Hadoop high-performance and reliable, in part because it allows for the data to be replicated across the cluster. The replication reduces the chance of data loss.

MapReduce is the default execution layer, which relies on a "divide-and-conquer" strategy to manage massive data sets and computing processes. Fully explaining MapReduce is beyond the scope of this document. However, the difficulty of developing real-world applications for the complex MapReduce framework is the original driving force behind the creation of Cascading.

As of Cascading 3.0, Apache Tez has become a third execution layer, or platform, option replacing MapReduce. See Hadoop 2 MapReduce vs Hadoop 2 Tez below.

Hadoop, according to its documentation, can be configured to run in three modes:

The pseudo-distributed mode does not add value for most purposes. This documentation does not explore this mode further.

Cascading itself can run in local mode or on the Hadoop platform, where Hadoop itself may be in stand-alone or distributed mode.

The primary difference between these two platforms, local or Hadoop, is that local-mode Cascading does not use Hadoop APIs and processes data in memory. In-memory processing quickens the runtime of applications, but consequently running on local mode is not as robust or scalable as running on the Hadoop platform.

Cascading supports both Hadoop 1.x and 2.x by providing two Java dependencies: cascading-hadoop.jar and cascading-hadoop2-mr1.jar. These dependencies can be interchanged but the hadoop2-mr1.jar introduces new APIs and deprecates older API calls where appropriate. It should be pointed out hadoop2-mr1.jar only supports MapReduce 1 API conventions. With this naming scheme new API conventions can be introduced without risk of naming collisions on dependencies.

Apache Hadoop 2 introduces YARN as the resource manager for the cluster. In short, this allows a single physical cluster to contain and execute different computing foundations simultaneously. By default a Hadoop 2 installation includes an implementation of MapReduce that is API-compatible with Apache Hadoop 1.

But YARN also allows other technologies to co-exist and share the same HDFS deployment. Apache Tez is one such technology. A single Hadoop YARN cluster can now execute some applications on MapReduce and others on Tez, where both frameworks could share or pass data via HDFS. The two resource managers can run concurrently without cluster-wide reconfiguration.

Apache Tez has yet to reach a 1.0 status, but is now currently supported by Cascading via the cascading-hadoop2-tez.jar dependency.

Since both MapReduce and Tez share HDFS, both implementations share much of the same APIs provided by Cascading. This greatly simplifies porting applications from MapReduce to Tez, in most cases the changes are to the build and to one line of code to switch to the appropriate FlowConnector.

The following sections describe the common features of each supported Hadoop-based platform.

See the Apache Tez Platform documentation of this Cascading User Guide for more information on running Apache Tez.

During runtime, Hadoop must be told which application JAR file to push to the cluster. The specific details are described in the platform-specific topics of this Cascading User Guide.

But to remain platform-independent, the AppProps class can be used, which is a helper fluent API for setting application-level configuration settings. Use this class for maximum portability.

Other Hadoop-specific Props classes include:

See the platform-specific topics of this Cascading User Guide for information about building an application for execution on a particular platform.

Running a Cascading application is the same as running any Hadoop application.

After packaging your application into a single JAR (see Building an Application), you must use bin/hadoop on Hadoop 1, or bin/yarn on Hadoop 2, to submit the application to the cluster.

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

If the configuration scripts in $HADOOP_CONF_DIR are configured to use a cluster, the JAR is pushed into that cluster for execution.

Cascading does not rely on any environment variables like $HADOOP_HOME or $HADOOP_CONF_DIR. Only the executable binary files in bin/hadoop and bin/yarn rely on those environment variables.

It should be noted that even though your-application.jar is passed on the command line to bin/hadoop or bin/yarn, this in no way configures Hadoop to push the 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, it’s likely that one of the internal libraries (found in the lib folder) will be pushed to the cluster instead, and "Class Not Found" exceptions will be thrown on the cluster.

Read about debugging and troubleshooting Cascading in local-mode troubleshooting before you start debugging.

For planner-related errors that present during runtime when executing a Flow, see The Cascading Process Planner.

Cascading provides a few HDFS-specific Tap and Scheme implementations. Regardless of the build dependencies, the package names remain consistent.

It is important to understand how Hadoop deals with directories. By default, Hadoop cannot source data from directories with nested subdirectories, and it cannot write to directories that already exist.

However, the good news is that you can simply point the Hfs tap (described below) to a directory of data files. The data files are all used as input. There is no need to enumerate each individual file into a MultiSourceTap. If there are nested directories, use GlobHfs.

You can integrate Cascading with either remote systems or file types by developing code that executes one of the following implementations:

Before beginning, or for inspiration, please see if any integrations for your file type or storage system already exist by visiting the Cascading extensions page.

In Hadoop mode, Cascading does not support MapReduce "Combiners." Combiners are a simple optimization allowing some Reduce functions to run on the Map side of MapReduce. Combiners are very powerful in that they reduce the I/O between the Mappers and Reducers — why send all of your Mapper data to Reducers when you can compute some values on the Map side and combine them in the Reducer?

But Combiners are limited to Associative and Commutative functions only, such as "sum" and "max." In addition, the process requires that the values emitted by the Map task must be serialized, sorted (which involves deserialization and comparison), deserialized again, and operated on — after which the results are again serialized and sorted. Combiners trade CPU for gains in I/O.

Cascading takes a different approach to partial aggregations on the Map side. When the results are combined on the Reduce side, Cascading trades memory instead of CPU for I/O gains. The Cascading mechanism functions by caching values (up to a threshold limit). This bypasses the redundant serialization, deserialization, and sorting. Also, Cascading allows any aggregate function to be implemented — not just Associative and Commutative functions.

Cascading supports any class type, except for an enum type, as values stored and passed in Tuple instances.

But for this to work when using the Cascading Hadoop mode, any Class that isn’t a primitive type or a Hadoop Writable type requires a corresponding Hadoop serialization class registered in the Hadoop configuration files for your cluster. Hadoop Writable types work because there is already a generic serialization implementation built into Hadoop. See the Hadoop documentation for information on registering a new serialization helper or creating Writable types.

Registered serialization implementations are automatically inherited by Cascading.

During serialization and deserialization of Tuple instances that contain custom types, the Cascading Tuple serialization framework must store the class name (as a String) before serializing the custom object. This can be very space-inefficient. To overcome this, custom types can add the SerializationToken Java annotation to the custom type class. The SerializationToken annotation expects two arrays — one of integers that are used as tokens, and one of Class name strings. Both arrays must be the same size. The integer tokens must all have values of 128 or greater because the first 128 values are reserved for internal use.

During serialization and deserialization, token values are used instead of the String Class names in order to reduce storage needs.

Serialization tokens may also be stored in the Hadoop configuration files. Alternatively, the tokens can be set as a property passed to the FlowConnector, with the property name cascading.serialization.tokens. The value of this property is a comma-separated list of token=classname values.

Note that Cascading natively serializes and deserializes all primitives and byte arrays (byte[]) if the developer registers the BytesSerialization class by using TupleSerializationProps.addSerialization( properties, BytesSerialization.class.getName() ). The token 127 is used for the Hadoop BytesWritable class.

By default, Cascading uses lazy deserialization on Tuple elements during comparisons when Hadoop sorts keys during the "shuffle" phase.

Cascading supports custom serialization for custom types, as well as lazy deserialization of custom types during comparisons. This is accomplished by implementing the StreamComparator interface. See the Javadoc for detailed instructions on implementation, and the unit tests for examples.