java operator api new functionality infosphere streams version 3.0

© 2012 IBM Corporation1

Java Operator APINew Functionality

InfoSphere Streams Version 3.0

Dan Debrunner and Howard NasgaardSPL Java Operator API


Important Disclaimer

THE INFORMATION CONTAINED IN THIS PRESENTATION IS PROVIDED FOR INFORMATIONAL PURPOSES ONLY.

WHILE EFFORTS WERE MADE TO VERIFY THE COMPLETENESS AND ACCURACY OF THE INFORMATION CONTAINED IN THIS PRESENTATION, IT IS PROVIDED “AS IS”, WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED.

IN ADDITION, THIS INFORMATION IS BASED ON IBM’S CURRENT PRODUCT PLANS AND STRATEGY, WHICH ARE SUBJECT TO CHANGE BY IBM WITHOUT NOTICE.

IBM SHALL NOT BE RESPONSIBLE FOR ANY DAMAGES ARISING OUT OF THE USE OF, OR OTHERWISE RELATED TO, THIS PRESENTATION OR ANY OTHER DOCUMENTATION.

NOTHING CONTAINED IN THIS PRESENTATION IS INTENDED TO, OR SHALL HAVE THE EFFECT OF:

• CREATING ANY WARRANTY OR REPRESENTATION FROM IBM (OR ITS AFFILIATES OR ITS OR THEIR SUPPLIERS AND/OR LICENSORS); OR

• ALTERING THE TERMS AND CONDITIONS OF THE APPLICABLE LICENSE AGREEMENT GOVERNING THE USE OF IBM SOFTWARE.

The information on the new product is intended to outline our general product direction and it should not be relied on in making a purchasing decision. The information on the new product is for informational purposes only and may not be incorporated into any contract. The information on the new product is not a commitment, promise, or legal obligation to deliver any material, code or functionality. The development, release, and timing of any features or functionality described for our products remains at our sole discretion.

THIS INFORMATION IS BASED ON IBM’S CURRENT PRODUCT PLANS AND STRATEGY, WHICH ARE SUBJECT TO CHANGE BY IBM WITHOUT NOTICE. IBM SHALL NOT BE RESPONSIBLE FOR ANY DAMAGES ARISING OUT OF THE USE OF, OR OTHERWISE RELATED TO, THIS PRESENTATION OR ANY OTHER DOCUMENTATION.


Agenda

SPL Compile Time Checking SPL Parameter handing ease-of-use SPL Trace & Log SPL Types Tuple Encoding Platform JMX Integration Minor Improvements SPADE API removal warning


SPL Compile Time Checking

Verify Java primitive operator invocation context at SPL compile time and/or runtime

Using Java code

– Invocation of annotated static methods of the operator's class at compile time

@ContextCheck

public static void checkFileDirectoryExclusive(OperatorContextChecker checker) {

checker.checkExcludedParameters("file", "directory");

}

– Including super-class methods

Javadoc: com.ibm.streams.operator.OperatorContext.ContextCheck


OperatorContextChecker

Passed into a @ContextCheck method Provides

– Access to compile time OperatorContext• Parameter names (values not available yet)• Compile time ports (e.g. no tuple submission)

– Get validity/set invalidity of invocation context

– Utility methods• similar to perl check helper functions in CodeGen.pm

public boolean checkDependentParameters(String parameterName, String ...dependentParameterNames);public boolean checkRequiredAttributes(StreamingData port, String ...attributeNames);

Javadoc: com.ibm.streams.operator.compile.OperatorContextChecker


How to cause SPL Compile Errors

Use OperatorContextChecker utility methods– If context invalid error SPL error automatically raised

Call OperatorContextChecker.setInvalidContext()– Typically with useful error message

Log an error or warn message– Error level causes an SPL compiler error– Warn level causes an SPL compiler warning

Throw an descriptive exception

Flexibility allows the Java primitive to raise multiple errors during compilation


Getting SPL Parameter values (Streams Version 2.0)

public class MyOp extends AbstractOperator {

private int threshold;

public void setThreshold(int threshold) { this.threshold = threshold; }

public int getThreshold() { return threshold; }

@Override

public void initialize(OperatorContext context) throws Exception {

super.initialize(context);

setThreshold(Integer.valueOf(

context.getParameterValues("threshold").get(0)));

}

}

Getting the value of a parameter is awkward:

- need to pull value out of List

- need to covert to specific type from String

Example is just an int, array values are harder

List<String> values = context.getParameterValues("rates");

double[] rates = new double[values.size()];

for (int i = 0; i < rates.length; i++)

rates[i] = Double.valueOf(values.get(i));

setRates(rates);


SPL Parameters mapped to setters (Streams Version 3.0)

Automatically map SPL parameters to the operator's Java bean properties

through an @Parameter annotation Javadoc: com.ibm.streams.operator.model.Parameter

Setter method for Java bean called automatically before initialize(), with:

- automatic conversion (with error checking)

- error checking for multiple values into a single value type (e.g. int)

- error checking for missing mandatory parameter

No code required at initialize() time - Removes boiler plate code

A bean property is just asetter method with astandard signature:

public void setName(type)

@Parameter

public void setThreshold(int threshold) {

this.threshold = threshold;

}

@Parameter(name=”flowRates”, optional=true)

public void setRates(double [] rates) {

this.rates = rates;

}


SPL Trace & Log

Support SPL trace & log enhancements– Integration with:

• Java platform java.util.logging

• Apache Log4j

– Existing trace Logging api deprecated

Loggers– Root (“”) map to SPL trace facility

• Thus arbitrary application loggers map to SPL trace

–com.ibm.streams.operator.log• map to SPL log facility

• Create a child logger for application messages to SPL log

Javadoc (package):– com.ibm.streams.operator.logging– com.ibm.streams.operator.log4j


Improved Trace/Log functionality

Can use all functionality of standard/log4j logging

– Child loggers– Localization (message bundles)– Trace/log to additional locations– Log levels– Multiple logging methods

SPL trace/log aspects supported by associating aspects with named loggers

Constants for SPL trace & log levels supplied– That map to native logger levels


LoggerUseExample (sample)public class LoggerUseExample extends AbstractOperator { private static final String CLASS_NAME = LoggerUseExample.class.getName();

/** Create a {@code Logger} specific to this class that will write * to the SPL trace facility as a child of the root {@code Logger}. */ private final Logger trace = Logger.getLogger(CLASS_NAME);

/** Create a {@code Logger} specific to this class that will write * to the SPL log facility as a child of the {@link LoggerNames#LOG_FACILITY} * {@code Logger}. * The {@code Logger} uses a resource bundle.*/ private final Logger log = Logger.getLogger(LoggerNames.LOG_FACILITY + "." + CLASS_NAME, "com.ibm.streams.operator.samples.operators.messages"); /** * Sample uses of log and trace. */ @Override public void initialize(OperatorContext context) throws Exception { super.initialize(context); // Note setLoggerAspects() must be called after super.initialize(context) // as it requires the context. // Associate the aspects exampleTrace and testTrace with trace // messages from the SPL trace logger setLoggerAspects(trace.getName(), "exampleTrace", "testTrace"); // Associate the aspect exampleLog with messages from the SPL log logger. setLoggerAspects(log.getName(), "exampleLog"); // Example use of trace to dump all parameters. The level uses // the Java Operator API specific class {@link TraceLevel} // that has constants for the SPL log levels. for (String parameterName : context.getParameterNames()) { for (String value : context.getParameterValues(parameterName)) { trace.log(TraceLevel.INFO, "Parameter {0} has value {1}", new Object[] {parameterName, value}); } }


SPL enum

enum type supported in stream schemas– com.ibm.streams.operator.Type.MetaType.ENUM

Java mapping– String – exact match to SPL identifier– Java enum

• State state =tuple.getEnum(State.class, “sensorstate”);

– SPL identifiers must match Java enum constants• Enum.name()


SPL XML

xml type supported in stream schemas– com.ibm.streams.operator.Type.MetaType.XML

Java mapping– com.ibm.streams.operator.types.XML

• Integrate with full Java XML support through– javax.xml.transform.stream.StreamSource

• Indication if value is default SPL value (“empty”)– ValueFactory.newXML() to create values

• Integrate with full Java XML support through– javax.xml.transform.Source– javax.xml.transform.Transformer

XML processing through standard Java apis– DOM, JAXB, SAX, StAX


Tuple Encoding

Serialize tuples to and from

– Java serialization (Streams Version 2.0)

– SPL native binary encoding

• Compatible with C++ NativeByteBuffer

Encode tuples to

– JSON (JavaScript Object Notation)

• Including collections & nested tuples

• Excluding blob, complex32, complex64

Javadoc (package):– com.ibm.streams.operator.encoding


JMX Integration - Life-cycle Notifications

JMX– Java Management & Monitoring API

– Rich set of functionality

MXBeans registered in platform MBeanServer– OperatorContextMXBean, ProcessingElementMXBean

– Supports operator life-cycle notifications• allPortsReady,shutdown,customMetricCreated

– Allows code to be shared across multiple operator classes and manage its own life-cycle

– E.g. Jetty webserver started by Operator.initialize() but watches allPortsReady & shutdown notifications to avoid requiring each operator class to code explicit notifications

Javadoc (package): – com.ibm.streams.operator.management


JMX Integration - Metrics

Ability to register a MetricMXBean for operator custom and port metrics

Allows integration of the JMX monitoring functionality provided by Java

– javax.management.monitoring

CounterMonitorMBean– Alert when threshold reached

GaugeMonitorMBean– Alerts when high and low threshold crossed


Input Port QueueJMX GaugeMonitorMBean Example

T2 T1 T0

Notification when queue firstcontains 8 tuples. No furthernotification until low thresholdis crossed

Notification when queue drops to 2 tuples after a high threshold notification

High threshold Low threshold

InputPortMetric.queueSizeInputPortMetric.nTuplesQueued

GaugeMonitorMBean

MetricMXBean

Allows an operator to change itsbehaviour based upon queue stateusing a hysteresis mechanism


Minor Improvements

OperatorContext

– getKind() - Get primitive operator kind

ProcessingElement

– Getters for applicationScope, applicationDirectory, optimized, relaunchCount, instanceId

StreamingData

– Mapping of operator port to PE port

Tuple,OutputTuple

– Better support for nested tuples

• AsReadOnlyTuple, setTuple, assignTuple


SPADE API Removal WARNING

com.ibm.streams.operator– Streams Version 2.0+ operator API– All new functionality extends this api only

com.ibm.streams.spade– Streams Version 1.x operator API– Deprecated in Streams Version 2.0

• Supported in Versions 2.0 & 3.0– TO BE REMOVED IN NEXT RELEASE

• Removal warnings added in Streams Version 3.0

java operator api new functionality infosphere streams version 3.0

Documents