Transforms

Mapping is the simplest kind of stateless transformation. It simply applies a function to the input item, and passes the output to the next stage.

Similar to map, the filter is a stateless operator that applies a predicate to the input to decide whether to pass it to the output.

flatMap is equivalent to map, with the difference that instead of one output item you can have arbitrary number of output items per input item. The output type is a Traverser, which is a Hazelcast type similar to an Iterator. For example, the code below will split a sentence into individual items consisting of words:

This transform merges the contents of two streaming sources into one. The item type in the right-hand stage must be the same or a subtype of the one in the left-hand stage. The items from both sides will be interleaved in arbitrary order.

This transform looks up each incoming item from the corresponding IMap and the result of the lookup is combined with the input item.

The above code can be thought of as equivalent to below, where the input is of type Order

See Joining Static Data to a Stream for a tutorial using this operator.

This transform is equivalent to mapUsingIMap with the only difference that a ReplicatedMap is used instead of an IMap.

This transform takes an input and performs a mapping using a service object. Examples are an external HTTP-based service or some library which is loaded and initialized during runtime (such as a machine learning model).

The service itself is defined through a ServiceFactory object. The main difference between this operator and a simple map is that the service is initialized once per job. This is what makes it useful for calling out to heavy-weight objects which are expensive to initialize (such as HTTP connections).

Let’s imagine an HTTP service which returns details for a product and that we have wrapped this service in a ProductService class:

We can then create a shared service factory as follows:

"Shared" means that the service is thread-safe and can be called from multiple-threads, so Hazelcast will create just one instance on each member and share it among the parallel tasklets.

We also declared the service as "non-cooperative" because it makes blocking HTTP calls. Failing to do this would have severe consequences for the performance of not just your pipeline, but all the jobs running on the Hazelcast cluster.

We can then perform a lookup on this service for each incoming order:

This transform is identical to mapUsingService with one important distinction: the service in this case supports asynchronous calls, which are compatible with cooperative concurrency and don’t need extra threads. It also means that we can have multiple requests in flight at the same time to maximize throughput. Instead of the mapped value, this transform expects the user to supply a CompletableFuture<T> as the return value, which will be completed at some later time.

For example, if we extend the previous ProductService as follows:

We still create the shared service factory as before:

The lookup instead becomes async, and note that the transform also expects you to return

The main advantage of using async communication is that we can have many invocations to the service in-flight at the same time which will result in better throughput.

This variant is very similar to the previous one, but instead of sending one request at a time, we can send in so-called "smart batches" (for a more in-depth look at the internals of the Jet engine, see Cooperative Multithreading). Hazelcast will automatically group items as they come, and allows to send requests in batches. This can be very efficient for example for a remote service, where instead of one roundtrip per request, you can send them in groups to maximize throughput. If we would extend our ProductService as follows:

We can then rewrite the transform as:

As you can see, there is some more code to write to combine the results back, but this should give better throughput given the service is able to efficient batching.

Hazelcast can call Python code to perform a mapping step in the pipeline. The prerequisite is that the Hazelcast servers are Linux or Mac with Python installed and that the hazelcast-jet-python module is deployed on the classpath, through being present in the lib directory. Hazelcast supports Python versions 3.5-3.7.

For a full tutorial, see Apply a Python Function.

You are expected to define a function, conventionally named transform_list(input_list), that takes a list of strings and returns a list of strings whose items match positionally one-to-one with the input list. Hazelcast will call this function with batches of items received by the Python mapping stage. If necessary, you can also use a custom name for the transforming function.

Internally Hazelcast launches Python processes that execute your function. It launches as many of them as requested by the localParallelism setting on the Python pipeline stage. It prepares a local virtual Python environment for the processes to run in and they communicate with it over the loopback network interface, using a bidirectional streaming gRPC call.

If you have some simple Python work that fits into a single file, you can tell Hazelcast just the name of that file, which is assumed to be a Python module file that declares transform_list:

And here’s an example of handler.py:

If you have an entire Python project that you want to use from Hazelcast, just name its base directory and Hazelcast will upload all it (recursively) to the cluster as a part of the submitted job. In this case you must also name the Python module that declares transform_list:

Normally your Python code will make use of non-standard libraries. Hazelcast recognizes the conventional requirements.txt file in your project’s base directory and will ensure all the listed requirements are satisfied.

Finally, Hazelcast also recognizes bash scripts init.sh and cleanup.sh. It will run those during the initialization and cleanup phases of the job. Regardless of the parallelism of the Python stage, these scripts run just once per job, and they run in the context of an already activated virtual environment.

One issue with making requirements.txt work is that in many production back-end environments the public internet is not available. To work around this you can pre-install all the requirements to the global (or user-local) Python environment on all Hazelcast servers. You can also take full control by writing your own logic in init.sh that installs the dependencies to the local virtual environment. For example, you can make use of pip --find_links.

hashJoin is a type of join where you have two or more inputs where all but one of the inputs must be small enough to fit in memory. You can consider a primary input which is accompanied by one or more side inputs which are small enough to fit in memory. The side inputs are joined to the primary input, which can be either a batch or streaming stage. The side inputs must be batch stages.

The last argument to hashJoin is a function that gets the input and the enriching item. Note that by default Hazelcast does an outer join: if the enriching stream lacks a given key, the corresponding function parameter will be null. You can request an inner join as well:

In this case the product argument is never null and if a given key is missing, the input Order item is filtered out.

Hazelcast also supports hash-joining with more streams at once through hashJoin2 and the hashJoinBuilder. Refer to their documentation for more details.

Data aggregation is the cornerstone of distributed stream processing. It computes an aggregate function (simple examples: sum or average) over the data items.

When used without a defined window, the aggregate() method applies a one-time aggregation over the whole of the input which is only possible in a bounded input (using BatchStage).

For example, a very simple aggregation will look like this:

This will output only one result, which is the count of all the items:

The Jet API provides several built in aggregation methods, such as:

For a complete list, please refer to the AggregateOperations class. You can also implement your own aggregate operations using the builder in AggregateOperation .

Typically you don’t want to aggregate all the items together, but group them by some key and then aggregate over each group separately. This is achieved by using the groupingKey transform and then applying an aggregation on it afterwards.

We can extend the previous example to group odd and even values separately:

Grouping is critical for aggregating massive data sets in distributed computing - otherwise you would not be able to make use of parallelization as effectively.

Rolling aggregation is similar to aggregate but instead of waiting to output until all items are received, it produces an output item for each input item. Because of this, it’s possible to use it in a streaming pipeline as well, as the aggregation is applied in a continuous way. The same pipeline from aggregate, can be rewritten to use a rollingAggregate transform instead:

Instead of a single line of output, we would get the following output instead:

The process of data aggregation takes a finite batch of data and produces a result. We can make it work with an infinite stream if we break up the stream into finite chunks. This is called windowing and it’s almost always defined in terms of a range of event timestamps (a time window).

Window transforms requires a stream which is annotated with timestamps, that is each input item has a timestamp associated with it. Timestamps are given in milliseconds and are generally represented in epoch format, as a simple long.

For a more in-depth look at the event time model, please refer to Event Time and Processing Time.

The general way to assign windows to a stream works as follows:

Tumbling windows are the most basic window type - a window of constant size that "tumbles" along the time axis. If you use a window size of 1 second, Hazelcast will group together all events that occur within the same second and you’ll get window results for intervals [0-1) seconds, then [1-2) seconds, and so on.

A simple example is given below:

When you run this pipeline, you will see output like this, where each output window is marked with start and end timestamps:

As with a normal aggregation, it’s also possible to apply a grouping to a windowed operation:

In this mode, the output would be keyed:

Sliding window is like a tumbling window that instead of hopping from one time range to another, slides along instead. It slides in discrete steps that are a fraction of the window’s length. If you use a window of size 1 second sliding by 100 milliseconds, Hazelcast will output window results for intervals [0.00-1.00) seconds, then [0.10-1.1) seconds, and so on.

We can modify the tumbling window example as below:

When you run this pipeline, you will see output like the following, where you can see that the start and end timestamps of the windows are overlapping.

Session window captures periods of activity followed by periods of inactivity. You define the "session timeout", i.e., the length of the inactive period that causes the window to close. The typical example of a session window is a user’s activity on a website (hence the name). There are bursts of activity (while the user is browsing website ) followed by rather long periods of inactivity.

As with other aggregate transforms, if you define a grouping key, there will be a separate, independent session window for each key.

In the example below, we want to find out how many different events each user had during a web session. The data source is a stream of events read from Kafka and we assume that the user session is closed after 15 minutes of inactivity:

If you had to allow a lot of event lateness, or if you just use large time windows, you may want to track the progress of a window while it is still accumulating events. You can order Hazelcast to give you, at regular intervals, the current status on all the windows it has some data for, but aren’t yet complete. For example, on the session window example we may want to get an update of all running session every second without waiting for the 15 minute timeout to get the full results:

The output of the windowing stage is in the form of KeyedWindowResult<String, Long>, where String is the word and Long is the frequency of the events in the given window. KeyedWindowResult also has an isEarly property that says whether the result is early or final.

The early results period works for all windows types. For example in a tumbling window, if you are working with a window size of one minute and there’s an additional 15-second allowed lateness for the late-coming events, this amounts to waiting up to 75 seconds from receiving a given event to getting the result it contributed to. Therefore, it may be desirable to ask Hazelcast to give us updates on the current progress every second.

Suppresses duplicate items from a stream. This operation applies primarily to batch streams, but also works on a windowed unbounded stream.

This example takes some input of integers and outputs only the distinct values:

We can also use distinct with grouping, but then two items mapping to the same key will be duplicates. For example the following will print only strings that have different first letters:

This is a possible output (Hazelcast can choose any of the names starting in "j"):

Sorting only works on a batch stage. It sorts the input according to its natural ordering (the type must be Comparable) or according to the Comparator you provide. Example:

This is how the output should look:

Hazelcast sorts the data in two steps: first it sorts the data that is local on each member, and then it merges the sorted streams into the final output. The first step requires O(n) heap memory, but the second step just receives the data in the correct order and has O(1) memory needs.

mapStateful is an extension of the simple map transform. It adds the capability to optionally retain mutable state.

The major use case of stateful mapping is recognizing a pattern in the event stream, such as matching start-transaction with end-transaction events based on an event correlation ID. More generally, you can implement any kind of state machine and detect patterns in an input of any complexity.

As with other stateful operations, you can also use a groupingKey to have a unique state per key.

For example, consider a pipeline that matches incoming TRANSACTION_START events to TRANSACTION_END events which can arrive unordered and when both are received outputs how long the transaction took.

This would be difficult to express in terms of a slidingWindow, because we can’t know how long a transaction would take in advance, and if it would span multiple windows. It can’t be expressed using sessionWindow either, because we don’t want to wait until the window times out before emitting the results.

Let’s say we have the following class:

We can then use the following mapStateful transform to match start and end events:

You will note that we also had to set an expiry time on the states (first parameter of the mapStateful method), otherwise would eventually run out of memory as we accumulate more and more transactions.

Co-grouping allows to join any number of inputs on a common key, which can be anything you can calculate from the input item. This makes it possible to correlate data from two or more different sources. In the same transform you are able to apply an aggregate function to all the grouped items.

As an example, we can use a sequence of events that would be typical on an e-commerce website: PageVisit and AddToCart. We want to find how many visits were required before an item was added to the cart. For simplicity, let’s say we’re working with historical data and we are processing this data from a set of logs.

After getting the two keyed streams, now we can join them:

This gives an item which contains the counts for both events for the same user id. From this, it’s easy to calculate the ratio of visits vs add to cart events.

Co-grouping can also be applied to windowed streams, and works exactly the same way as aggregate(). An important consideration is that the timestamps from both streams would be considered, so it’s important that the two streams don’t have widely different timestamps.