Striim

TQL applications may import Java functions. The basic procedure is:

Functions are loaded at the cluster level, so your new function would be available for import to any application in any namespace. Imported functions required by an application are included as IMPORT statements in exported TQL.

The rest of this section describes in detail how to create and use custom functions.

// You will need your own package name:
package com.mycompany.custom.packagename;  

 
// You'll need to reference this is a Striim class in your own code if you
// define an Aggregate Window function:
import com.webaction.runtime.compiler.custom.AggHandlerDesc;

// Logging classes required by Striim  
import org.apache.log4j.Level;
import org.apache.log4j.Logger;
  
// Any other classes you may need for your development
import java.lang.Math;
import java.util.LinkedList;
import org.joda.time.DateTime;

// Abstract class needed to contain your custom functions
public abstract class MyCustomFunctions {

  // Required logger
  private static Logger logger = Logger.getLogger( MyCustomFunctions.class );
}

public static double maxSingleRow(double... n) {
  int    i=0;
  double runningmax = n[i];
 
  // Iterate through the row and continually update runningmax to the highest value
  while (++i < n.length)
    if (n[i] > runningmax)
      runningmax = n[i];
 
  return runningmax;
}

USE StriimWindows_NS2;
DROP APPLICATION StriimWindows2 CASCADE;
 
IMPORT STATIC com.mycompany.custom.packagename.MyCustomFunctions.*;
 
CREATE APPLICATION StriimWindows2;
 
CREATE OR REPLACE CQ myStream_CQ
INSERT INTO myStream_ST
SELECT
      maxSingleRow( field1, field2, field3, field4 )
FROM
      inputStream_ST;

	public static double minSingleRow(double... n) {
	    int i = 0;
	    double runningmin = n[i];
	    while (++i < n.length)
	        if (n[i] < runningmin)
	        	runningmin = n[i];
	    return runningmin;
	}	
	
	public static double rangeSingleRow(double... n) {
	    double dRange = maxSingleRow( n ) - minSingleRow( n );
    	return dRange;
	}	

	public static double avgSingleRow(double... n) {
	    int i = 0;
	    double runningsum = 0;
	    while (i < n.length) {
	        runningsum += n[i];
	        i++;
	    }
	    return ( runningsum / i );
	}	

    // Required handler declaration:
    // handler name must match the name of your static class. 
    @AggHandlerDesc(handler=AvgDebug_Float.class)

    // Name of the function to be exported to TQL:
    public abstract Float AvgDebug( String sKey, Float fArg );

    // Handler class implementing the required accumulators:
    // must match the name in the handler declaration.
    public static class AvgDebug_Float
    {
        // initializers
        float fRunningSum = 0;
        int iRunningCount = 0;
        
        // Calculates and returns the average value at any moment
        public Float getAggValue()
        {
            logger.debug( "RunningSum: " + fRunningSum + ",
              RunningCount: " + iRunningCount + "\n");
            return (iRunningCount == 0) ? 0 : (fRunningSum / iRunningCount);
        }
        // Adds the new value to the running total, maintaining the count
        // of values included:
        public void incAggValue(String sKey, Float fArg)
        {
            if(fArg != null) {
            	iRunningCount++;
                fRunningSum += fArg;
            }
            logger.debug( "[Key: " + sKey + ", Value: " +  fArg + "]
              RunningSum is now: " + fRunningSum + ",  RunningCount is now: "
              + iRunningCount );
        }

        // Removes the specified value from the running total, maintaining
        // the count of values included:
        public void decAggValue(String sKey, Float fArg)
        {
            if(fArg != null) {
            	iRunningCount--;
                fRunningSum -= fArg;
            }
            logger.debug( "[Key: " + sKey + ", Value: " +  fArg + "]
              RunningSum is now: " + fRunningSum + ",  RunningCount is now: " 
              + iRunningCount );
        }
    }

USE StriimWindows_NS2;
DROP APPLICATION StriimWindows2 CASCADE;
 
IMPORT STATIC com.mycompany.custom.packagename.MyCustomFunctions.*;
 
CREATE APPLICATION StriimWindows2;
 
CREATE OR REPLACE CQ MovingAvg_CQ
INSERT INTO MovingAvg_ST
SELECT
     avgDebug( OrderId, OrderValue )
FROM
     MovingAvg3ES_WN;

	@AggHandlerDesc(handler=LastBut_String.class)
    public abstract String LastBut(String arg, int idx);
    public static class LastBut_String
    {
        LinkedList<String> list = new LinkedList<String>();
        int i;
        
        // Returns the string located in the position just before 
        // the last specified index:
        public String getAggValue() 
        { 
            if( list.isEmpty()  || (i < 0) || (i > (list.size()-1)) ) { 
                return null;
            } else {
                return list.get( ( list.size()-i-1) );
            }
        }
 
        // Adds a string to the end of the linked list, 
        //updating the index location:
        public void incAggValue(String arg, int idx) 
        { 
            list.addLast(arg);
            i = idx;  // no safety logic required here - 
                      // all handled by the getAggValue() method
        }


        // Removes the string located at the head of the linked list,
        // assuming it is not an empty list:
        public void decAggValue(String arg, int idx) 
        { 
            if(!list.isEmpty()) list.removeFirst();
        }
    }

    @AggHandlerDesc(handler=LastBut_Byte.class)
    public abstract Byte LastBut(Byte arg, int idx);
    public static class LastBut_Byte
    {
        LinkedList<Byte> list = new LinkedList<Byte>();
        int i;
        
        public Byte getAggValue() 
        { 
            if( list.isEmpty()  || (i < 0) || (i > (list.size()-1)) ) { 
                return null;
            } else {
                return list.get( ( list.size()-i-1) );
            }
        }
        public void incAggValue(Byte arg, int idx) 
        { 
            list.addLast(arg);
            i = idx;  // no safety logic required here - 
                      // all handled by the getAggValue() method
        }
        public void decAggValue(Byte arg, int idx) 
        { 
            if(!list.isEmpty()) list.removeFirst();
        }
    }

SELECT
  AVG(OrderValue),
  OrderID,
  OrderDate,
  OrderValue
FROM
  STREAM_ST

log4j.logger.com.mycompany.custom.packagename.MyCustomFunctions=debug, MyCustomFunctionsAppender
log4j.additivity.com.mycompany.custom.packagename.MyCustomFunctions=false
log4j.appender.MyCustomFunctionsAppender=org.apache.log4j.FileAppender
log4j.appender.MyCustomFunctionsAppender.File=logs/MyCustomFunctions.log.out
# Define the layout for file appender
log4j.appender.MyCustomFunctionsAppender.layout=org.apache.log4j.EnhancedPatternLayout
log4j.appender.MyCustomFunctionsAppender.layout.conversionPattern=%-6p:%-15M:%m%n

    @AggHandlerDesc(handler=AvgDebug_Float.class)
    public abstract Float AvgDebug( String sKey, Float fArg );
    public static class AvgDebug_Float
    {
        float fRunningSum = 0;
        int iRunningCount = 0;
        
        public Float getAggValue()
        {
            logger.debug( "RunningSum: " + fRunningSum + ",  RunningCount: " + iRunningCount + "\n");
            return (iRunningCount == 0) ? 0 : (fRunningSum / iRunningCount);
        }
        public void incAggValue(String sKey, Float fArg)
        {
            if(fArg != null) {
            	iRunningCount++;
                fRunningSum += fArg;
            }
            logger.debug( "[Key: " + sKey + ", Value: " +  fArg + "] RunningSum is now:
              " + fRunningSum + ",  RunningCount is now: " + iRunningCount );
        }
        public void decAggValue(String sKey, Float fArg)
        {
            if(fArg != null) {
            	iRunningCount--;
                fRunningSum -= fArg;
            }
            logger.debug( "[Key: " + sKey + ", Value: " +  fArg + "] RunningSum is now:
              " + fRunningSum + ",  RunningCount is now: " + iRunningCount );
        }
    }

set loglevel = {com.mycompany.custom.packagename.MyCustomFunctions : debug};

import com.webaction.runtime.compiler.custom.Nondeterministic; 
...

@Nondeterministic
public static int kickShell (String args) {
        int ret = 0;
        try {
            if (args.indexOf(".sh") == -1) {
                System.out.println("Error:" + args + " is not shell.");
                return ret;
            }
            Process process = new ProcessBuilder("sh", args).start();
            ret = process.waitFor();
            System.out.println("return:" + ret);
        }catch (ArrayIndexOutOfBoundsException e) {
            System.out.println("Info: Please specify one argument.");
            System.out.println("return:" + ret);
        }catch (Exception e){
            System.out.println("return:" + ret);
            e.printStackTrace();
        }finally{
            return ret;
        }
    }

CREATE CQ ...
SELECT kickShell('/etc/striim/test.sh') as retVal ...

Functions are loaded at the cluster level, so are available for import to any application in any namespace. Imported functions are included as IMPORT statements in exported TQL. The LOAD and UNLOAD commands require the Global.admin role.

Before loading your custom function, copy its .jar file to striim/lib. Whenever Striim is restarted, the function will be loaded automatically.

To load the function using the console, enter LOAD "lib/<file name>.jar";

To import a loaded function in the console or a TQL file, use IMPORT STATIC <package name>.

To load and import a custom function in the web UI, select App Settings > Add Another Package, enter IMPORT STATIC <package name> in the Java Package field, and click Save.

To unload a custom function, in the console enter UNLOAD "lib/<file name>.jar";

To stop the function from reloading when Striim is restarted, delete its .jar file from striim/lib.

A Striim open processor contains a custom Java application that reads data from a window or stream, processes it, optionally enriching it with data from a cache, and writes to an output stream.

The SDK includes the following:

The component must be built with Maven, since it requires the Maven Shade Plugin.

An open processor can be used only in the Striim namespace from which the types are exported.

The following simple example shows all the steps required to create an open processor and use it in a Striim application.

Step 1: define the input and output streams in Striim

The following TQL defines the input and output streams for the example open processor you will add later. It includes a FileWriter source, a cache that will be specified in the open processor's ENRICH option, and a FileWriter target.

CREATE NAMESPACE ns1;
USE ns1;
CREATE APPLICATION OPExample;

CREATE source CsvDataSource USING FileReader (
  directory:'Samples/PosApp/appData',
  wildcard:'PosDataPreview.csv',
  positionByEOF:false
)
PARSE USING DSVParser (
  header:Yes,
  trimquote:false
)
OUTPUT TO CsvStream;
 
CREATE TYPE MerchantHourlyAve(
  merchantId String,
  hourValue integer,
  hourlyAve integer
);

CREATE CACHE HourlyAveLookup using FileReader (
  directory: 'Samples/PosApp/appData',
  wildcard: 'hourlyData.txt'
)
PARSE USING DSVParser (
  header: Yes,
  trimquote:false,
  trimwhitespace:true
) 
QUERY (keytomap:'merchantId') 
OF MerchantHourlyAve;

CREATE CQ CsvToPosData
INSERT INTO PosDataStream partition by merchantId
SELECT TO_STRING(data[1]) as merchantId,
  TO_DATEF(data[4],'yyyyMMddHHmmss') as dateTime,
  DHOURS(TO_DATEF(data[4],'yyyyMMddHHmmss')) as hourValue,
  TO_DOUBLE(data[7]) as amount,
  TO_INT(data[9]) as zip
FROM CsvStream;
 
CREATE CQ cq2
INSERT INTO SendToOPStream
SELECT makeList(dateTime) as dateTime,
  makeList(zip) as zip
FROM PosDataStream;
 
CREATE TYPE ReturnFromOPStream_Type ( time DateTime , val Integer );
CREATE STREAM ReturnFromOPStream OF ReturnFromOPStream_Type;

CREATE TARGET OPExampleTarget 
USING FileWriter (filename: 'OPExampleOut') 
FORMAT USING JSONFormatter() 
INPUT FROM ReturnFromOPStream;
 
END APPLICATION OPExample;

Step 2: export the input and output stream types

If you create OPExample in the ns1 workspace, the following Striim console command will export the types from the application to /home/myhome/OpExampleTypes.jar:

EXPORT TYPES OF ns1.OPExample TO "/home/myhome/OpExampleTypes.jar";

The EXPORT TYPES command requires read permission on the namespace. If you do not specify a path, the file will be created in the striim directory.

Step 3: set up Maven

Install the SDK and exported types .jar files:

mvn install:install-file -DgroupId=com.example -DartifactId=OpenProcessorSDK \
  -Dversion=1.0.0-SNAPSHOT -Dpackaging=jar -Dfile=/opt/striim/StriimSDK/StriimOpenProcessor-SDK.jar
mvn install:install-file -DgroupId=com.example -DartifactId=OPExample -Dversion=1.0.0-SNAPSHOT \
  -Dpackaging=jar -Dfile=/home/myhome/OpExampleTypes.jar

Create a Maven project in which you will create your custom Java application:

mvn archetype:generate -DgroupId=com.example.opexample -DartifactId=opexample \
  -DarchetypeArtifactId=maven-archetype-quickstart -DinteractiveMode=false

Replace the default pom.xml created by Maven with the following, adjusting as necessary for your environment:

<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
  xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
  <modelVersion>4.0.0</modelVersion>
  <groupId>com.example.opexample</groupId>
  <artifactId>opexample</artifactId>
  <packaging>jar</packaging>
  <version>1.0-SNAPSHOT</version>
  <name>opexample</name>
  <url>http://maven.apache.org</url>
  <dependencies>
    <dependency>
      <groupId>junit</groupId>
      <artifactId>junit</artifactId>
      <version>3.8.1</version>
      <scope>test</scope>
    </dependency>
<!-- OpenProcessorSDK jar -->
        <dependency>
            <groupId>com.example</groupId>
            <artifactId>OpenProcessorSDK</artifactId>
            <version>1.0.0-SNAPSHOT</version>
            <scope>provided</scope>
        </dependency>
<!-- exported types jar -->
        <dependency>
            <groupId>com.example</groupId>
            <artifactId>OPExample</artifactId>
            <version>1.0.0-SNAPSHOT</version>
            <scope>provided</scope>
        </dependency>
  </dependencies>
    <build>
        <plugins>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-release-plugin</artifactId>
                <version>2.2.2</version>
            </plugin>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-compiler-plugin</artifactId>
                <version>2.3.2</version>
                <configuration>
                    <source>1.8</source>
                    <target>1.8</target>
                </configuration>
            </plugin>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-shade-plugin</artifactId>
                <version>2.4.3</version>
                <configuration>
                    <createDependencyReducedPom>false</createDependencyReducedPom>
<!-- 
The output SCM filename is defined here.
-->
                    <finalName>OPExample.scm</finalName>
                    <transformers>
<transformer implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer">
                            <manifestEntries>
                                <Striim-Module-Name>OPExample</Striim-Module-Name>
                                <Striim-Service-Interface>
                                  com.webaction.runtime.components.openprocessor.StriimOpenProcessor
                                </Striim-Service-Interface>
                                <Striim-Service-Implementation>
                                  com.example.opexample.App
                                </Striim-Service-Implementation>
                            </manifestEntries>
                        </transformer>
                    </transformers>
                    <artifactSet>
                        <excludes>
                            <exclude>org.slf4j:*</exclude>
                            <exlcude>log4j:*</exlcude>
                        </excludes>
                    </artifactSet>
                </configuration>
                <executions>
                    <execution>
                        <phase>package</phase>
                        <goals>
                            <goal>shade</goal>
                        </goals>
                    </execution>
                </executions>
            </plugin>
            <plugin>
                <groupId>com.coderplus.maven.plugins</groupId>
                <artifactId>copy-rename-maven-plugin</artifactId>
                <version>1.0</version>
                <executions>
                    <execution>
                        <id>copy-file</id>
                        <phase>package</phase>
                        <goals>
                            <goal>copy</goal>
                        </goals>
<!--
The location and name for the .scm file to be imported into Striim is defined here.
Preferred location is module/modules folder under the Maven project main folder.
-->
                        <configuration>
    <sourceFile>/home/myhome/opexample/target/OpExample.scm.jar</sourceFile>
    <destinationFile>/home/myhome/opexample/modules/OpExample.scm</destinationFile>
                        </configuration>
                    </execution>
                </executions>
            </plugin>
        </plugins>
    </build>
</project>

Step 4: write your Java application and build the .scm

Replace the default App.java with the following:

package com.example.opexample;
 
import wa.ns1.SendToOPStream_Type_1_0;
import wa.ns1.ReturnFromOPStream_Type_1_0;
 
import com.webaction.anno.PropertyTemplateProperty;
import com.webaction.runtime.components.openprocessor.StriimOpenProcessor;
import org.joda.time.DateTime;
 
import com.webaction.anno.AdapterType;
import com.webaction.anno.PropertyTemplate;
import com.webaction.runtime.containers.WAEvent;
 
import com.webaction.runtime.containers.IBatch;
import java.util.*;
 
@PropertyTemplate(name = "TupleConverter", type = AdapterType.process,
properties = {
@PropertyTemplateProperty(name="ahead", type=Integer.class, required=true, defaultValue="0"),
@PropertyTemplateProperty(name="lastItemSeen", type=Boolean.class, required=true, defaultValue="0")
},
// The names of outputType and inputType are relative to Striim: output from a native Striim
// code to your custom component, and input from your custom component to a native component.
outputType = SendToOPStream_Type_1_0.class,
inputType = ReturnFromOPStream_Type_1_0.class
)
public class App extends StriimOpenProcessor
{
  
public void run() {
		IBatch<WAEvent> event = getAdded();
		Iterator<WAEvent> it = event.iterator();
		while (it.hasNext()) {
			SendToOPStream_Type_1_0  type = (SendToOPStream_Type_1_0 ) it.next().data;
			//  ... Additional operations
		}

		ReturnFromOPStream_Type_1_0 ReturnFromOPStream_Type_1_0  = new ReturnFromOPStream_Type_1_0 ();
		ReturnFromOPStream_Type_1_0.time = DateTime.now();
		Random rand = new Random(System.currentTimeMillis());

		ReturnFromOPStream_Type_1_0.val= rand.nextInt(50) + 1;
		send(ReturnFromOPStream_Type_1_0 );

	}

public void close() throws Exception {
        // TODO Auto-generated method stub
 
    }
 
    public Map getAggVec() {
        // TODO Auto-generated method stub
        return null;
    }
 
    public void setAggVec(Map aggVec) {
        // TODO Auto-generated method stub
 
    }
}

Change to the opexample directory created by Maven and enter mvn package.

Step 5: import the .scm into Striim

Loading an open processor requires the Global.admin permission (see Permissions).

Copy opexample/modules/OpExample.scm to a directory accessible by the Striim server, then use the following console command to load it:

LOAD OPEN PROCESSOR "<path>/OpExample.scm";

Alternatively, you may load it in Flow Designer at Configuration > App settings > Load / unload open processor.

In either case, when the application is restarted, Striim will reload the open processor from the same location.

Step 6: add the open processor to your application

Return to the application you created in step 1, open the Base Components section of the component palette, drag a Striim Open Processor into the workspace, set its options as follows, and click Save. Note that Ahead and Last Item Seen are defined by the Java class. The other properties will appear in all open processor components.

If you run the application, it will create output files in the striim directory.

Modifying an open processor

To modify your open processor, unload it in Flow Designer at Configuration > App settings > Load / unload open processor or using the command UNLOAD OPEN PROCESSOR "<path>/<file name>.scm";. Then make changes to the Java application, compile, and load the new .scm.

The LOAD and UNLOAD commands require the Global.admin role.

To load an open processor using the console, enter: LOAD OPEN PROCESSOR "<path>/<file name>.scm";

To load an open processor in the Flow Designer, select App Settings, enter <path>/<file name>.scm in the Load/Unload Open Processor field, and click Load.

To unload an open processor using the console, enter: UNLOAD "<path>/<file name>.scm";

To unload an open processor in the Flow Designer, select App Settings, enter <path>/<file name>.scm in the Load/Unload Open Processor field, and click Unload.