Groups incoming records according to record conditions instead of fixed time windows. The aggregation starts when readings meet a predicate, continues while they qualify, and forwards results when the condition ends or the aggregate reaches its configured limit. section 10.3
Monitoring manufacturing temperatures
A Waldo Widgets application ignores temperatures below a threshold, aggregates consecutive elevated readings, and emits useful summaries instead of forwarding every sensor event. A reading below the threshold ends the active aggregation. section 10.3
A Processor API implementation separates the aggregation responsibilities:
- Predicates decide whether a sensor reading starts or continues aggregation and when it should stop. section 10.3
- A key-value state store retains the current SensorAggregation for each record key. section 10.3
- A scheduled punctuation scans stored aggregates, removes stale dangling entries, and forwards them. section 10.3
Listing 10.10 ProcessorSupplier
public class DataDrivenAggregate implements
ProcessorSupplier<String, Sensor, String, SensorAggregation> {
private final StoreBuilder<?> storeBuilder; #1
private final Predicate<Sensor> shouldAggregate; #2
private final Predicate<Sensor> stopAggregation; #3
public DataDrivenAggregate(final StoreBuilder<?> storeBuilder,
final Predicate<Sensor> shouldAggregate,
final Predicate<Sensor> stopAggregation) {
this.storeBuilder = storeBuilder;
this.shouldAggregate = shouldAggregate;
this.stopAggregation = stopAggregation;
}
Inspect the Java ProcessorSupplier wiring the state-store builder and the predicates used by the custom processor. section 10.3
