Custom missing value processor
Build, deploy and apply a custom transformer
Objective
We will create a simple missing data transformer that fills in missing values by either adding a default value or using the previous field's value.
Prerequisites
To get started building a custom processor ensure you have your development environment configured.
Read the environment setup documentation to get your environment ready to build.
Development steps
These instructions cover how to build, deploy a use the processor on to the Joule Platform.
Create project using the template
We have provided a project template project to quick start development. The project can be found here. Clone the template project and copy relevant code and structure to your own project.
git clone [email protected]:joule-platform/fractalworks-project-templates.git
Implement missing value transformer
Processors differ from connectors as they do not require, currently, a specification and builder classes. So jump right in and create and name a class that reflects the processing function.
Joule provides the core logic such as batching, cloning, linking of data stores, and a unique processor UUID for event change lineage.
Key areas of implementation:
Define processor DSL namespace
Initialize and apply methods
Attribute setters and properties
Add the class definition to plugins.properties
Deploy and apply to a Joule runtime environment
Code implementation
package com.yourcompany.processor.transformers;
import com.fasterxml.jackson.annotation.JsonRootName;
import com.fasterxml.jackson.annotation.JsonProperty;
import com.fractalworks.streams.core.data.streams.Context;
import com.fractalworks.streams.core.data.streams.StreamEvent;
import com.fractalworks.streams.core.data.streams.Metric;
import java.util.HashMap;
import java.util.Map;
// JsonRootName value will be used in the use case definition
@JsonRootName(value = "missing value transformer")
public class CustomMissingValueTransformer extends AbstractProcessor {
private Map<String,Object> previousValue = new HashMap<>();
private String key;
private String field;
private Object defaultValue;
public CustomMissingValueTransformer() {
super();
}
@Override
public void initialize(Properties prop) throws ProcessorException {
super.initialize(prop);
// Add specific initialisation code here
}
/**
* This is were your custom code is provided
*/
@Override
public StreamEvent apply(StreamEvent event, Context context)
throws StreamsException {
var value = event.getValue(field);
if(value == null){
value = (previousValue.containsKey(key)
? previousValue.getValue(previousValue): defaultValue;
event.addValue(uuid,field,value);
if (logger.isInfoEnabled()) {
logger.info("Updated missing {} value with {}",field,value);
}
}
previousValue.put(key, value);
// JMX enabled metrics
metrics.incrementMetric(Metric.PROCESSED);
return event;
}
/**
* Attribute setters and dsl property
*/
@JsonProperty(value = "unique key", required = true)
public void setField(String key) {
this.key = key;
}
@JsonProperty(value = "field", required = true)
public void setField(String field) {
this.field = field;
}
@JsonProperty(value = "default value", required = true)
public void setDefaultValue(Object defaultValue) {
this.defaultValue = defaultValue;
}
}
Note: If you would like to perform batch processing override the below method.
public MicroBatch apply(MicroBatch batch, Context context) throws StreamsException;
Deploy
Once your package has been successfully created you are ready to deploy to a Joule project.
The resulting jar from the build process needs to be copied to the userlibs
directory under a Joule project directory. For example using the getting started project copy the file to quickstart/userlibs
directory.
cp build/libs/<your-processor>.jar <location>/userlibs
Now apply to a stream
Lets say, sometimes we do not get a bid value which is needed to trigger an alert. So overcome a division by zero we provide a default value and use previous values when needed.
stream:
name: nasdaq_major_banks_stream
eventTimeType: EVENT_TIME
processing unit:
pipeline:
# Filter events by major banks
- filter:
expression: "(typeof industry !== 'undefined' &&
industry == 'Major Banks')"
# Apply
- missing value transformer:
key: symbol
field: bid
default value: 1.0
emit:
select: symbol, bid, ask
# Spread trigger
having: "((bid - ask) / bid) > 0.015"
group by:
- symbol
Follow the same steps used in the getting started documentation to apply this script.
What we have learnt
As a first process we have covered a number of key features:
Build a simple transformer Used the provided template project to quick start development and add custom code within key processor methods.
Built the jar Used gradle build tool to build, test and deploy to local maven repo.
Deploy the jar to a Joule runtime environment Copied the Jar to an existing local Joule runtime environment
Apply transformer within a use case Apply the transformer within a use case to provide consistent spread alerts in the event of missing data.
Last updated
Was this helpful?