Multiple Composition and Execution
After the current graph has been executed, a new
LogicalGraph is created and becomes the current graph. At this point the current graph can again be composed and executed. This process can be repeated as necessary.
he following RushScript example demonstrates using a for loop to repeatedly compose and execute an application. The script:
1. Creates a schema for the input data.
2. Constructs a JavaScript array containing the names of the input files to process.
3. Uses a for loop to handle each file name in the array.
• Within the for loop, a DataFlow application is composed and executed. The application simply reads the data file for this iteration, performs the aggregation, and collects the aggregation results.
4. The java.lang.System class is used to write log messages to standard output. The class is imported into the JavaScript environment in the first line of the script.
Repeatedly composing and executing an application
importClass(Packages.java.lang.System);
var schema = dr.schema()
.STRING("eventID")
.TIMESTAMP("timestamp")
.STRING("siteID")
.INT("compromised")
.LONG("entityID");
var files = new Array("logs1.txt", "logs2.txt", "logs3.txt");
for (i = 0; i < files.length; i++) {
var logData = dr.readDelimitedText({source:files[i], schema:schema, fieldSeparator:"|"});
var results = dr.group(logData, {aggregations: Aggregation.sum("compromised").as("compromised")});
var collector = results.collectResults();
dr.execute("Log loop " + i);
var data = collector.getData();
var compromisedCount = data[0].get('compromised');
System.out.println(files[i] + ": compromised count: " + compromisedCount);
}
The captured output from the execution of the example RushScript is shown below. The output shows that the DataFlow application was executed three times. After each execution the file name processed and the total count of compromised fields are printed.
INFO com.pervasive.dataflow.graphs.internal.LogicalGraphInstanceImpl execute Executing phase 0 graph: {[AssignSplits, ParseSplits, ReducePartialAggregates, SortedPartialAggregator, collector_output]}
INFO com.pervasive.dataflow.operators.io.LocalityOptimizer next Assigned 1 total splits across 8 nodes, with 1 non-local
com.pervasive.dataflow.graphs.internal.LogicalGraphInstanceImpl execute Phase 0 graph: {[AssignSplits, ParseSplits, ReducePartialAggregates, SortedPartialAggregator, collector_output]} completed in 695 milliseconds
INFO com.pervasive.dataflow.script.javascript.DataflowFactory execute script execution time: 0.88 secs
logs1.txt: compromised count: 710
INFO com.pervasive.dataflow.graphs.internal.LogicalGraphInstanceImpl execute Executing phase 0 graph: {[AssignSplits, ParseSplits, ReducePartialAggregates, SortedPartialAggregator, collector_output]}
INFO com.pervasive.dataflow.operators.io.LocalityOptimizer next Assigned 1 total splits across 8 nodes, with 1 non-local
INFO com.pervasive.dataflow.graphs.internal.LogicalGraphInstanceImpl execute Phase 0 graph: {[AssignSplits, ParseSplits, ReducePartialAggregates, SortedPartialAggregator, collector_output]} completed in 226 milliseconds
INFO com.pervasive.dataflow.script.javascript.DataflowFactory execute script execution time: 0.24 secs
logs2.txt: compromised count: 710
INFO com.pervasive.dataflow.graphs.internal.LogicalGraphInstanceImpl execute Executing phase 0 graph: {[AssignSplits, ParseSplits, ReducePartialAggregates, SortedPartialAggregator, collector_output]}
INFO com.pervasive.dataflow.operators.io.LocalityOptimizer next Assigned 1 total splits across 8 nodes, with 1 non-local
INFO com.pervasive.dataflow.graphs.internal.LogicalGraphInstanceImpl execute Phase 0 graph: {[AssignSplits, ParseSplits, ReducePartialAggregates, SortedPartialAggregator, collector_output]} completed in 173 milliseconds
INFO com.pervasive.dataflow.script.javascript.DataflowFactory execute script execution time: 0.19 secs
logs3.txt: compromised count: 710