Development of an Apex application starts with mapping the functional specification to operators (smaller functional building blocks), which can then be composed into a DAG to collectively provide the functionality required for the use case.

This involves identifying the data sources, formats, transformations and sinks for the application, and finding matching operators from the Apex library (which will be covered in the next chapter). In most cases, the required connectors will be available from the library that support frequent sources, such as files and Kafka, along with many other external systems that are part of the Apache Big Data ecosystem.

With the comprehensive operator library and set of examples to cover frequently used I/O cases and transformations, it is often possible to assemble a preliminary end-to-end flow that covers a subset of the functionality quickly, before building out the complete business logic in detail.

Development of Apex applications requires a Java development environment with the following:

In addition to the above, it is recommended to have an IDE with Maven support such as IntelliJ or Eclipse. Apex provides code style settings for these IDEs that can optionally be used.

It is further recommended to have Git installed. Git is not required to build an application, but it is a convenient way to fetch the Apex source code and is especially useful for easily navigating the full operator library (apex-malhar) project within the IDE when working on operator customizations.

Apex applications are packaged in a special ZIP file format that contains everything needed for an application to be launched on a cluster (dependency jars, configuration files, and so on). It is roughly comparable to the uber jar approach that some other frameworks employ, with the difference that dependencies in the Apex package remain as individual JAR files, rather than being flattened into a standard JAR.

It would be a rather involved task to set up a new Maven project from scratch. The Apex application archetype simplifies the process of creating an application skeleton for the expected artifact structure. Here is an example of the Maven command to create an Apex application archetype:

mvn archetype:generate \
  -DarchetypeGroupId=org.apache.apex \
  -DarchetypeArtifactId=apex-app-archetype -DarchetypeVersion=RELEASE...

Let's start by transforming this placeholder application into an application that counts words – the Hello World equivalent for big data processing frameworks. The functionality is easy to understand and not very important, as our focus here is on the development process.

The full source code of the modified application is available at https://github.com/tweise/apex-samples/tree/master/wordcount. Here is the modified application assembly in Application.java:

@Override
   public void populateDAG(DAG dag, Configuration conf)
   {
    LineByLineFileInputOperator lineReader = dag.addOperator("input",
         new LineByLineFileInputOperator());
     LineSplitter parser = dag.addOperator("parser", new LineSplitter());
     UniqueCounter counter = dag.addOperator("counter", new UniqueCounter());
     GenericFileOutputOperator<Object> output = dag.addOperator("output",
         new GenericFileOutputOperator<>());
     output.setConverter(new ToStringConverter...

As our example application has the LineSplitter operator, which is not part of the Apex library, we will use it as an example to illustrate the process of developing a custom operator.

Splitting a line into words is, of course, a simple stateless operation. Connectors and stateful transformations will be more involved, and there are many examples in the Apex library to look at for this.

Here is the line splitter:

public class LineSplitter extends BaseOperator 
{ 
  // default pattern for word-separators 
  private static final Pattern nonWordDefault =    Pattern.compile
    ("[\\p{Punct}\\s]+"); 
 
  private String nonWordStr;              // configurable regex 
  private transient Pattern nonWord;      // compiled regex 
 
  /** 
   * Output port on which words from the current file are emitted 
   */ 
  public final transient DefaultOutputPort<String> output = new 
    DefaultOutputPort<>(); 
 
  /** 
   * Input port on which lines from the current...