Architecture

PDR uses a dataflow architecture. The architecture implements a program that executes as a graph of computation nodes interconnected by dataflow queues. The nodes use the queues to share data. In this sense, dataflow is a shared-nothing architecture. The lack of share state simplifies node implementation, since threads do not have to synchronize share state. The in-memory, blocking queues implement the synchronization required to safely hand off data from node to node.

In PDR, the computation nodes of a dataflow graph are known as operators. PDR provides a library of ready-to-use operator components. You can also write custom operators to extend the standard library. For example, several of the sample applications have their own implementations of operators.

To support the creation of a dataflow graph for execution, PDR provides a composition phase for constructing operators and linking them in an execution graph. Operator properties can be set to determine both operator composition and runtime behavior. At runtime, a composed graph is realized by creating threads for each computation node, creating dataflow queues, and linking nodes. The Pervasive DataRush execution engine also supports monitoring using JMX. During the execution phase, statistics objects may be created and MBeans instantiated to export profile and debug information. Pervasive DataRush provides a VisualVM plug-in that can be used within VisualVM to display the exported run-time information.

The next two topics give more detail about composition and execution.

Composition

Pervasive DataRush supports two types of operator, DataflowGraph and DataflowNode, both Java interfaces. DataflowGraph is a composite operator, used only to compose other operators. After composition, a DataflowGraph operator no longer exists (it is "compiled away"). DataflowNode is an executable operator attached to a thread and executed at run time. Both operator types can be used to set operator properties and can be linked by dataflow queues.

Dataflow queues are not instantiated at composition time to prevent premature access before run time. During composition, operators are linked using a flow concept. When an operator is composed, its internal structure is created and its methods exposed for obtaining output flows. The output flow of one operator can be passed as input to another operator, which uses the passed-in flow to complete linking.

ApplicationGraph is a special DataflowGraph, used to create an application to run within the Pervasive DataRush engine. Like other graphs, it lives at composition time and has an interface for adding operators. Once composed, the ApplicationGraph can be run. You can use the static method on the OperatorFactory to obtain an implementation of an ApplicationGraph.

Execution

After a graph is composed, it is ready to run. The ApplicationGraph interface defines a run method. On instantiation, engine properties set during composition define monitoring structures. At run time, threads are launched and the main thread then waits for either normal thread completion or an error.

The PDR engine includes a deadlock algorithm that is instantiated whenever a thread has to wait on a queue according to certain criteria. The algorithm looks for cycles in the wait graph. If any are found, then deadlock has occurred. Without intervention, graph execution halts while the deadlock algorithm determines which queue is at fault and expands memory for that queue. Deadlocks are thus often transient and occur only occasionally on a graph under particular stress.

Monitoring

The execution of a PDR application can be monitored using VisualVM, the JMX console that is shipped with the Java JDK. Pervasive DataRush ships with a plugin for VisualVM. The plugin can be found in the plugins directory under the PDR installation. It is contained in the file named datarush-visualvm-*.nbm. Follow the instructions within VisualVM for installing a new plugin. To obtain runtime information, connect to the JVM executing Pervasive DataRush using VisualVM. You can use the Pervasive DataRush tabs within VisualVM to see the running nodes, view queue information, and obtain general JVM and system information.

Return to Top

Writing a Pervasive DataRush Application

Join Sample

The PDR release includes several sample applications located in the samples directory. The samples include the source code with sample input data. We'll use the join sample for this tutorial. The purpose of the join sample is to read in two, pre-sorted text files, join them together and output the results of the join to a third file. This is a very simple example, but is useful to demonstrate several concepts within PDR.

Source code:

public class JoinSample extends DataflowGraphBase
{
    private static final String[] JSON_PROPERTY_KEYS = {
        "leftInputProperties", "rightInputProperties", "joinKeys", "outputProperties"
    };

    public static void main(String[] args) {
        // Parse command-line flags, and read and parse properties file
        CliConfig cli = new CliParser().parseArgs(args);
        cli.parse(JSON_PROPERTY_KEYS);
        // Reflectively construct a configured JoinSample, and run it
        cli.run(JoinSample.class);
    }

    public JoinSample (
            @Param("leftInputFile") String leftInputFile,
            @Param("leftInputProperties") DelimitedTextProperties leftInputProperties,
            @Param("rightInputFile") String rightInputFile,
            @Param("rightInputProperties") DelimitedTextProperties rightInputProperties,
            @Param("joinMode") JoinMode joinMode,
            @Param("joinKeys") String[] keyNames,
            @Param("outputFile") String outputFile,
            @Param("outputProperties") DelimitedTextProperties outputProperties) {

        // Add reader for left hand side of join
        ReadDelimitedText readLeft = add(
                new ReadDelimitedText(leftInputFile, leftInputProperties), "readLeft");

        // Add reader for right hand side of join
        ReadDelimitedText readRight = add(
                new ReadDelimitedText(rightInputFile, rightInputProperties), "readRight");

        // Add join
        JoinSortedRows join = add(
                new JoinSortedRows(readLeft.getOutput(), readRight.getOutput(), keyNames, keyNames, joinMode),
                "join");

        // Add writer for output of join
        add(new WriteDelimitedText(join.getOutput(), outputFile, outputProperties), "write");
    }
}

Note that JoinSample extends DataflowGraphBase, instead of using ApplicationGraph. This sample uses the PDR CLIParser and CLIConfig helper classes and so does not need to be an ApplicationGraph. The CLIConfig class exposes a run method that is invoked within the main of JoinSample. The CLIParser is used to parse the arguments passed to the main method. One of the standard options (-pf) can be used to pass in a property file. The CLIParser maps between the properties within the given properties file, by name, to the annotations on the JoinSample constructor. For example, if the property file contains a property named "leftInputFile", the value for that property will be passed to the JoinSample constructor as the leftInputFile parameter. Javascript Object Notation (JSON) is used to help construct complex properties such as lists, maps and Java objects.

A DataflowGraph that exposes a main and uses the CLI helper classes to process arguments can be invoked directly using the PDR dr script (found in the bin directory). The dr script builds up the command line needed to invoke the JVM to run the wanted application. For example, to run the join application, the following command line can be used:

dr example.join.JoinSample -pf join.properties

The composition of the join application happens in the JoinSample constructor. Since JoinSample extends DataflowGraphBase, it has exposed to it the add method. This method takes either a DataflowGraph or a DataflowNode to add to the composition of JoinSample. Looking at the constructor code, we can see that it first creates and then adds a ReadDelimitedText operator.This operator reads a text file in delimited form and outputs a dataflow containing the data from the file. Another ReadDelimitedText operator is constructed and added to the graph. This one reads the second file to be joined. Next, the JoinSortedRows operator is created and added to the graph. This operators does the join of the data. Note that we are assuming the two input data sources are already joined. The first two parameters of the join operator constructor are the two dataflows to join together. The join operator is passed the output of the first text reader and the output of the second text reader. In this way, the operators are linked together. Next, a WriteDelimitedText operator is created and added to the graph. This operator reads from a dataflow and writes to a delimited text file. Note that the writer takes the output dataflow of the join as its input. The writer does not produce a dataflow output.

Now, back to the main method. The CLIParser is used to parse the arguments passed to the class by the invocation from the JVM (which was invoked by the dr script). The parser finds the -pf option, parses the property file and creates a CliConfig object. The class to invoke (JoinSample) and the parsed config are passed to the run method. The run method creates an ApplicationGraph, constructs the JoinSample and adds it to the application graph, and then executes the application. The run method does not return until the JoinSample application completes execution.

The DataflowGraph implementations are composite operators, including ApplicationGraph. They can contain composite operators that may contain composite operators and so on. We can see this behavior in the above example: a DataflowGraph is created and the JoinSample is added to it. JoinSample contains several composite operators which themselves may use other composites. 

When the JoinSample graph is executed, the readers will commence to reading and parsing their input files and stuffing data onto their output queues. In parallel, the join operator will read its two input data queues, join the incoming data by the specified keys fields and output the data to its output queue. The writer will read data from the join operator and write the data to an output text file. All of the components are able to run in independent threads in pipe-lined fashion, hence taking advantage of multiple CPUs in parallel.

Return to Top

IDEs

Eclipse Integration