eventStreamProcessor

Apply stream analytic function to event stream

Since R2022b

This object requires Streaming Data Framework for MATLAB^® Production Server™.

Description

Use an EventStreamProcessor object to apply a stream analytic function to an event stream. Using EventStreamProcessor object functions, you can automatically direct events from an event stream to a streaming analytic function, enabling you to process large amounts of data in event streams.

You can run the streaming analytic function on a known number of event windows synchronously, similar to a for-loop. You can also run it with a desktop-hosted server to simulate asynchronous deployment in a production environment.

EventStreamProcessor functions can process streaming data sequentially in batches by collecting events into windows of configurable size. When a window is full of the requested number of events, the window of events is passed to the stream processing analytic function. You can then save any results that the analytic function produces and optionally publish them to a different stream.

A stream processing function can be stateful or stateless. For stateful functions, the EventStreamProcessor object maintains state between calls to the stream processing function. If the stream processing function changes the state, the function can return the state as a second output argument. The EventStreamProcessor object preserves these changes for the next function iteration.

Creation

Syntax

esp = eventStreamProcessor(inputStream,streamFcn)

esp = eventStreamProcessor(inputStream,streamFcn,initialState)

esp = eventStreamProcessor(___,Name=Value)

Description

esp = eventStreamProcessor(inputStream,streamFcn) creates an EventStreamProcessor object, which applies the stream function streamFcn to the event stream inputStream, and sets the InputStream and StreamFunction properties, respectively, of this object.

esp = eventStreamProcessor(inputStream,streamFcn,initialState) creates an EventStreamProcessor object that additionally initializes persistent state with the function initialState and sets the InitialState property. If streamFcn is stateful, then initialState is required.

esp = eventStreamProcessor(___,Name=Value) sets object properties using one or more name-value arguments. Name is a property name and Value is the corresponding value. You can specify multiple name-value arguments in any order as Name1=Value1,...,NameN=ValueN.

Properties

ArchiveName — Name of generated deployable archive

string array

Name of the deployable archive generated by the package function, specified as a string. The default archive name is the name of the streaming function.

Data Types: string

GroupVariable — Name of event variable used to group events

string array | character vector

Name of the event variable used to group events, specified as a string array or character vector.

If GroupVariable is nonempty, each event window is split into groups in which the event variables have the same value. Each group is then sent to the streaming function separately. GroupVariable is often set to the event key so that events from each event source are processed independently.

Data Types: string | char

InitialState — Function that creates initial state for streaming analytic function

function handle

Function that creates the initial state for the streaming analytic function, specified as a function handle. If the streaming analytic function is stateful, this property must be set when you create the object.

InputStream — Event stream from which the streaming analytic function reads events

KafkaStream object | InMemoryStream object | TestStream object

Event stream from which the streaming analytic function reads events, specified as a KafkaStream, InMemoryStream, or TestStream object.

OutputStream — Event stream to which streaming analytic function writes events

InMemoryStream object (default) | KafkaStream object | TestStream object

Event stream to which the streaming analytic function writes events, specified as a KafkaStream, InMemoryStream, or TestStream object.

Note

If you are packaging your stream processing function into a deployable archive using the package function, do not leave OutputStream set to an InMemoryStream object. This object is not supported by the package function as an output stream.

StateStore — Persistent storage connection name

string | character vector

Persistent storage connection name, specified as a string or character vector. You must specify a StateStore when using InitialState or when using a stateful stream function. The connection name must be known to the MATLAB Production Server instance to which the archive will be deployed. For more information on using a data cache for persistent storage, see Data Caching Basics.

Data Types: char | string

StreamFunction — Streaming analytic function

function handle

Streaming analytic function, specified as a function handle.

Data Types: function handle

ReadPosition — Position in event stream to read from

"Beginning" | "End" | "Current"

Position in an event stream to read from, specified as one of these values:

• "Beginning" — First event available in stream

• "End" — Just past the last event in the stream

• "Current" — Just past the current event in the stream

Data Types: string

ResetStateOnSeek — Flag to clear persistent state after calling seek

true (default) | false

Flag to clear persistent state after calling the seek function, specified as a logical scalar.

Data Types: logical

Object Functions

execute	Execute event stream processing function on specific number of event windows
package	Package stream processing function into deployable archive configured by EventStreamProcessor
seek	Set position in event stream to begin processing events
start	Start processing event streams using local test server
startServer	Start local test server
stop	Stop processing event streams using local test server
stopServer	Shut down local test server

Examples

Iterate Streaming Analytic Function Over Several Event Windows

Assume that you have a Kafka^® server running at the network address kafka.host.com:9092 that has a topic RecamanSequence.

Create an object connected to the RecamanSequence topic.

ks = kafkaStream("kafka.host.com",9092,"RecamanSequence");

Assume that you have a streaming analytic function recamanSum and a function to initialize persistent state called initRecamanSum.

Create an EventStreamProcessor object to run the recamanSum function and initializes persistent state with the initRecamanSum function.

esp = eventStreamProcessor(ks,@recamanSum,@initRecamanSum);

esp = 

  EventStreamProcessor with properties:

      StreamFunction: @recamanSum
         InputStream: [1×1 matlab.io.stream.event.KafkaStream]
        OutputStream: [1×1 matlab.io.stream.event.InMemoryStream]
        InitialState: @initRecamanSum
       GroupVariable: [0×0 string]
        ReadPosition: Beginning
         ArchiveName: "recamanSum"
    ResetStateOnSeek: 1

Iterate the streaming analytic function over the stream ten times.

execute(esp,10);

Examine the results.

result = readtimetable(esp.OutputStream)

Simulate Production Using Test Server

Assume that you have a Kafka server running at the network address kafka.host.com:9092 that has a topic RecamanSequence.

Also assume that you have a streaming analytic function recamanSum and a function initRecamanSum to initialize persistent state.

Create a KafkaStream object connected to the RecamanSequence topic.

ks = kafkaStream("kafka.host.com",9092,"RecamanSequence");

Create another KafkaStream object to write the results of the streaming analytic function to a different topic called RecamanSequenceResults.

outKS = kafkaStream("kafka.host.com",9092,"RecamanSequenceResults");

Create an EventStreamProcessor object that runs the recamanSum function and initializes persistent state with the initRecamanSum function.

esp = eventStreamProcessor(ks,@recamanSum,@initRecamanSum,OutputStream=outKS);

esp = 

  EventStreamProcessor with properties:

      StreamFunction: @recamanSum
         InputStream: [1×1 matlab.io.stream.event.KafkaStream]
        OutputStream: [1×1 matlab.io.stream.event.KafkaStream]
        InitialState: @initRecamanSum
       GroupVariable: [0×0 string]
        ReadPosition: Beginning
         ArchiveName: "recamanSum"
    ResetStateOnSeek: 1

Using the MATLAB editor, you can set breakpoints in the recamanSum function to examine the incoming streaming data when you start the server.

Start the test server.

Note

To use the test server, you require MATLAB Compiler SDK™.

startServer(esp);

Doing so opens the Production Server Compiler app. When the app opens, you must start the server manually.

To start the test server from the app, click Test Client, and then click Start. For an example on how to use the app, see Test Client Data Integration Against MATLAB (MATLAB Compiler SDK).

Navigate back to the MATLAB command prompt to start processing events.

start(esp);

In the Production Server Compiler app, the test server receives data.

From the MATLAB editor, if you set breakpoints, you can use the debugger to examine the data, state, and results of the function processing. Click Continue to continue debugging or Stop when you finish debugging.

From the MATLAB command prompt, stop the server.

stop(esp);

Read results from the output stream.

results = readtimetable(outKS);

Version History

Introduced in R2022b