Application Performance

import java.util.List;
import com.pervasive.datarush.graphs.EngineConfig;
import com.pervasive.datarush.graphs.InputPortInstance;
import com.pervasive.datarush.graphs.LogicalGraph;
import com.pervasive.datarush.graphs.LogicalGraphFactory;
import com.pervasive.datarush.graphs.LogicalGraphInstance;
import com.pervasive.datarush.graphs.LogicalStatistic;
import com.pervasive.datarush.graphs.OperatorInstance;
import com.pervasive.datarush.operators.LogicalOperator;
import com.pervasive.datarush.operators.sink.LogRows;
import com.pervasive.datarush.operators.source.GenerateRandom;
import com.pervasive.datarush.ports.LogicalPort;
import com.pervasive.datarush.types.RecordTokenType;
import com.pervasive.datarush.types.TokenTypeConstant;

/**
* Create a simple application graph consisting of a random row generator and
* a row logger. Execute the graph with the default engine properties.
*
*/
public class MonitoringAppInstance {
    public static void main(String[] args) {

        // Create an empty logical graph
        LogicalGraph graph = LogicalGraphFactory.newLogicalGraph("SimpleApp");

        // Create a random row generator setting the row count and type wanted
        GenerateRandom generator = graph.add(new GenerateRandom(), "generator");
        generator.setRowCount(100000000);
        RecordTokenType type = TokenTypeConstant.record(
                TokenTypeConstant.DOUBLE("dblField"),
                TokenTypeConstant.STRING("stringFld"));
        generator.setOutputType(type);

        // Create a row logger
        LogRows logger = graph.add(new LogRows(0), "logger");

        // Connect the output of the generator to the input of the logger
        graph.connect(generator.getOutput(), logger.getInput());

        // Compile the graph with monitoring enabled
        LogicalGraphInstance instance = graph.compile(EngineConfig.engine().monitored(true));

        // Start the instance. This call is asyncrhonous. It returns right away.
        // Use join to wait for graph completion.
        instance.start();

        // Monitor the graph while it is running.
        do {
            // Dump stats
            dumpOpStats(instance, generator, logger);
            dumpCxnStats(instance, logger.getInput());

            // Sleep and then check the instance state
            try {
                Thread.sleep(1000);
            }
            catch (InterruptedException e) {
                break;
            }

        } while (!instance.getState().isComplete());

        dumpOpStats(instance, generator, logger);
        dumpCxnStats(instance, logger.getInput());

        // Join with the graph instance to clean up
        try {
            instance.join();
        }
        catch (InterruptedException e) {
            e.printStackTrace();
        }

    }

    private static void dumpOpStats(LogicalGraphInstance graphInstance, LogicalOperator... ops) {
        for (LogicalOperator op : ops) {
            OperatorInstance opInstance = graphInstance.getOperatorInstance(op);
            System.out.println(opInstance.getPath() + ": state: " + opInstance.getState());
            dumpStats(opInstance.getStatistics());
        }
    }

    private static void dumpCxnStats(LogicalGraphInstance instance, LogicalPort...ports) {
        for (LogicalPort port : ports) {
            InputPortInstance portInstance = instance.getPortInstance(port);
            dumpStats(portInstance.getStatistics());
        }
    }

    private static void dumpStats(List<LogicalStatistic> stats) {
        for (LogicalStatistic stat : stats) {
            System.out.println(stat.path() + ": partition "+stat.nodeID()+ ": "+stat.definition()+": "+stat.supplier());
        }
    }
}