org.apache.samza.operators

Interface MessageStream<M>

Type Parameters:

M - the type of messages in this stream

@InterfaceStability.Evolving
public interface MessageStream<M>

A stream of messages that can be transformed into another MessageStream.

A MessageStream corresponding to an input stream can be obtained using StreamApplicationDescriptor.getInputStream(org.apache.samza.system.descriptors.InputDescriptor<M, ?>).

Method Summary

Modifier and Type	Method and Description
MessageStream<M>	broadcast(Serde<M> serde, java.lang.String id) Broadcasts messages in this MessageStream to all instances of its downstream operators..
MessageStream<M>	filter(FilterFunction<? super M> filterFn) Applies the provided function to messages in this MessageStream and returns the filtered MessageStream.
<OM> MessageStream<OM>	flatMap(FlatMapFunction<? super M,? extends OM> flatMapFn) Applies the provided 1:n function to transform a message in this MessageStream to n messages in the transformed MessageStream
<OM> MessageStream<OM>	flatMapAsync(AsyncFlatMapFunction<? super M,? extends OM> asyncFlatMapFn) Applies the provided 1:n transformation asynchronously to this MessageStream.
<K,OM,JM> MessageStream<JM>	join(MessageStream<OM> otherStream, JoinFunction<? extends K,? super M,? super OM,? extends JM> joinFn, Serde<K> keySerde, Serde<M> messageSerde, Serde<OM> otherMessageSerde, java.time.Duration ttl, java.lang.String id) Joins this MessageStream with another MessageStream using the provided pairwise JoinFunction.
<K,R extends KV,JM> MessageStream<JM>	join(Table<R> table, StreamTableJoinFunction<? extends K,? super M,? super R,? extends JM> joinFn, java.lang.Object... args) Joins this MessageStream with another Table using the provided pairwise StreamTableJoinFunction.
<OM> MessageStream<OM>	map(MapFunction<? super M,? extends OM> mapFn) Applies the provided 1:1 function to messages in this MessageStream and returns the transformed MessageStream.
MessageStream<M>	merge(java.util.Collection<? extends MessageStream<? extends M>> otherStreams) Merges all otherStreams with this MessageStream.
static <T> MessageStream<T>	mergeAll(java.util.Collection<? extends MessageStream<? extends T>> streams) Merges all streams.
<K,V> MessageStream<KV<K,V>>	partitionBy(MapFunction<? super M,? extends K> keyExtractor, MapFunction<? super M,? extends V> valueExtractor, KVSerde<K,V> serde, java.lang.String id) Re-partitions this MessageStream using keys from the keyExtractor by creating a new intermediate stream on the default system provided via ApplicationDescriptor.withDefaultSystem(org.apache.samza.system.descriptors.SystemDescriptor<?>).
MessageStream<M>	sendTo(OutputStream<M> outputStream) Allows sending messages in this MessageStream to an OutputStream and then propagates this MessageStream to the next chained operator
<K,V> MessageStream<KV<K,V>>	sendTo(Table<KV<K,V>> table, java.lang.Object... args) Allows sending messages in this MessageStream to a Table and then propagates this MessageStream to the next chained operator.
void	sink(SinkFunction<? super M> sinkFn) Allows sending messages in this MessageStream to an output system using the provided SinkFunction.
<K,WV> MessageStream<WindowPane<K,WV>>	window(Window<M,K,WV> window, java.lang.String id) Groups the messages in this MessageStream according to the provided Window semantics (e.g.

Method Detail

map

<OM> MessageStream<OM> map(MapFunction<? super M,? extends OM> mapFn)

Applies the provided 1:1 function to messages in this MessageStream and returns the transformed MessageStream.

Type Parameters:

OM - the type of messages in the transformed MessageStream

Parameters:

mapFn - the function to transform a message to another message

Returns:

the transformed MessageStream

flatMap

<OM> MessageStream<OM> flatMap(FlatMapFunction<? super M,? extends OM> flatMapFn)

Applies the provided 1:n function to transform a message in this MessageStream to n messages in the transformed MessageStream

Type Parameters:

OM - the type of messages in the transformed MessageStream

Parameters:

flatMapFn - the function to transform a message to zero or more messages

Returns:

the transformed MessageStream

flatMapAsync

<OM> MessageStream<OM> flatMapAsync(AsyncFlatMapFunction<? super M,? extends OM> asyncFlatMapFn)

Applies the provided 1:n transformation asynchronously to this MessageStream. The asynchronous transformation is specified through AsyncFlatMapFunction. The results are emitted to the downstream operators upon the completion of the CompletionStage returned from the AsyncFlatMapFunction.

The operator can operate in two modes depending on task.max.concurrency..

• Serialized (task.max.concurrency=1) - In this mode, each invocation of the AsyncFlatMapFunction is guaranteed to happen-before next invocation.
• Parallel (task.max.concurrency>1) - In this mode, multiple invocations can happen in parallel without happens-before guarantee and the AsyncFlatMapFunction is required synchronize any shared state. The operator doesn't provide any ordering guarantees. i.e The results corresponding to each invocation of this operator might not be emitted in the same order as invocations. By extension, the operator chain that follows it also doesn't have any ordering guarantees.

Type Parameters:

OM - the type of messages in the transformed MessageStream

Parameters:

asyncFlatMapFn - the async function to transform a message to zero or more messages

Returns:

the transformed MessageStream

filter

MessageStream<M> filter(FilterFunction<? super M> filterFn)

Applies the provided function to messages in this MessageStream and returns the filtered MessageStream.

The FilterFunction is a predicate which determines whether a message in this MessageStream should be retained in the filtered MessageStream.

Parameters:

filterFn - the predicate to filter messages from this MessageStream.

Returns:

the filtered MessageStream

sink

void sink(SinkFunction<? super M> sinkFn)

Allows sending messages in this MessageStream to an output system using the provided SinkFunction.

Offers more control over processing and sending messages than sendTo(OutputStream) since the SinkFunction has access to the MessageCollector and TaskCoordinator. This can also be used to send output to a system (e.g. a database) that doesn't have a corresponding Samza SystemProducer implementation.

Parameters:

sinkFn - the function to send messages in this stream to an external system

sendTo

MessageStream<M> sendTo(OutputStream<M> outputStream)

Allows sending messages in this MessageStream to an OutputStream and then propagates this MessageStream to the next chained operator

When sending messages to an OutputStream<KV<K, V>>, messages are partitioned using their serialized key. When sending messages to any other OutputStream<M>, messages are partitioned using a null partition key.

Note: The message will be written but not flushed to the underlying output system before its propagated to the chained operators. Messages retain the original partitioning scheme when propogated to next operator.

Parameters:

outputStream - the output stream to send messages to

Returns:

this MessageStream

window

<K,WV> MessageStream<WindowPane<K,WV>> window(Window<M,K,WV> window,
                                              java.lang.String id)

Groups the messages in this MessageStream according to the provided Window semantics (e.g. tumbling, sliding or session windows) and returns the transformed MessageStream of WindowPanes.

Use the Windows helper methods to create the appropriate windows.

The id must be unique for each operator in this application. It is used as part of the unique ID for any state stores and streams created by this operator (the full ID also contains the job name, job id and operator type). If the application logic is changed, this ID must be reused in the new operator to retain state from the previous version, and changed for the new operator to discard the state from the previous version.

Type Parameters:

K - the type of key in the message in this MessageStream. If a key is specified, panes are emitted per-key.

WV - the type of value in the WindowPane in the transformed MessageStream

Parameters:

window - the window to group and process messages from this MessageStream

id - the unique id of this operator in this application

Returns:

the windowed MessageStream

join

<K,OM,JM> MessageStream<JM> join(MessageStream<OM> otherStream,
                                 JoinFunction<? extends K,? super M,? super OM,? extends JM> joinFn,
                                 Serde<K> keySerde,
                                 Serde<M> messageSerde,
                                 Serde<OM> otherMessageSerde,
                                 java.time.Duration ttl,
                                 java.lang.String id)

Joins this MessageStream with another MessageStream using the provided pairwise JoinFunction.

Messages in each stream are retained for the provided ttl and join results are emitted as matches are found.

Both inputs being joined must have the same number of partitions, and should be partitioned by the join key.

The id must be unique for each operator in this application. It is used as part of the unique ID for any state stores and streams created by this operator (the full ID also contains the job name, job id and operator type). If the application logic is changed, this ID must be reused in the new operator to retain state from the previous version, and changed for the new operator to discard the state from the previous version.

Type Parameters:

K - the type of join key

OM - the type of messages in the other stream

JM - the type of messages resulting from the joinFn

Parameters:

otherStream - the other MessageStream to be joined with

joinFn - the function to join messages from this and the other MessageStream

keySerde - the serde for the join key

messageSerde - the serde for messages in this stream

otherMessageSerde - the serde for messages in the other stream

ttl - the ttl for messages in each stream

id - the unique id of this operator in this application

Returns:

the joined MessageStream

join

<K,R extends KV,JM> MessageStream<JM> join(Table<R> table,
                                           StreamTableJoinFunction<? extends K,? super M,? super R,? extends JM> joinFn,
                                           java.lang.Object... args)

Joins this MessageStream with another Table using the provided pairwise StreamTableJoinFunction.

The type of input message is expected to be KV.

Records are looked up from the joined table using the join key, join function is applied and join results are emitted as matches are found.

The join function allows implementation of both inner and left outer join. A null will be passed to the join function, if no record matching the join key is found in the table. The join function can choose to return an instance of JM (outer left join) or null (inner join); if null is returned, it won't be processed further.

Both the input stream and table being joined must have the same number of partitions, and should be partitioned by the same join key.

Type Parameters:

K - the type of join key

R - the type of table record

JM - the type of messages resulting from the joinFn

Parameters:

table - the table being joined

joinFn - the join function

args - additional arguments passed to the table

Returns:

the joined MessageStream

merge

MessageStream<M> merge(java.util.Collection<? extends MessageStream<? extends M>> otherStreams)

Merges all otherStreams with this MessageStream.

The merged stream contains messages from all streams in the order they arrive.

Parameters:

otherStreams - other MessageStreams to be merged with this MessageStream

Returns:

the merged MessageStream

mergeAll

static <T> MessageStream<T> mergeAll(java.util.Collection<? extends MessageStream<? extends T>> streams)

Merges all streams.

The merged MessageStream contains messages from all streams in the order they arrive.

Type Parameters:

T - the type of messages in each of the streams

Parameters:

streams - MessageStreams to be merged

Returns:

the merged MessageStream

Throws:

java.lang.IllegalArgumentException - if streams is empty

partitionBy

<K,V> MessageStream<KV<K,V>> partitionBy(MapFunction<? super M,? extends K> keyExtractor,
                                         MapFunction<? super M,? extends V> valueExtractor,
                                         KVSerde<K,V> serde,
                                         java.lang.String id)

Re-partitions this MessageStream using keys from the keyExtractor by creating a new intermediate stream on the default system provided via ApplicationDescriptor.withDefaultSystem(org.apache.samza.system.descriptors.SystemDescriptor<?>). This intermediate stream is both an output and input to the job.

Uses the provided KVSerde for serialization of keys and values.

The number of partitions for this intermediate stream is determined as follows: If the stream is an eventual input to a join(org.apache.samza.operators.MessageStream<OM>, org.apache.samza.operators.functions.JoinFunction<? extends K, ? super M, ? super OM, ? extends JM>, org.apache.samza.serializers.Serde<K>, org.apache.samza.serializers.Serde<M>, org.apache.samza.serializers.Serde<OM>, java.time.Duration, java.lang.String), and the number of partitions for the other stream is known, then number of partitions for this stream is set to the number of partitions in the other input stream. Else, the number of partitions is set to the value of the job.intermediate.stream.partitions configuration, if present. Else, the number of partitions is set to to the max of number of partitions for all input and output streams (excluding intermediate streams).

The id must be unique for each operator in this application. It is used as part of the unique ID for any state stores and streams created by this operator (the full ID also contains the job name, job id and operator type). If the application logic is changed, this ID must be reused in the new operator to retain state from the previous version, and changed for the new operator to discard the state from the previous version.

Unlike sendTo(org.apache.samza.operators.OutputStream<M>), messages with a null key are all sent to partition 0.

Type Parameters:

K - the type of output key

V - the type of output value

Parameters:

keyExtractor - the MapFunction to extract the message and partition key from the input message. Messages with a null key are all sent to partition 0.

valueExtractor - the MapFunction to extract the value from the input message

serde - the KVSerde to use for (de)serializing the key and value.

id - the unique id of this operator in this application

Returns:

the repartitioned MessageStream

sendTo

<K,V> MessageStream<KV<K,V>> sendTo(Table<KV<K,V>> table,
                                    java.lang.Object... args)

Allows sending messages in this MessageStream to a Table and then propagates this MessageStream to the next chained operator. The type of input message is expected to be KV, otherwise a ClassCastException will be thrown.

Note: The message will be written but may not be flushed to the underlying table before its propagated to the chained operators. Whether the message can be read back from the Table in the chained operator depends on whether it was flushed and whether the Table offers read after write consistency. Messages retain the original partitioning scheme when propogated to next operator.

Type Parameters:

K - the type of key in the table

V - the type of record value in the table

Parameters:

table - the table to write messages to

args - additional arguments passed to the table

Returns:

this MessageStream

broadcast

MessageStream<M> broadcast(Serde<M> serde,
                           java.lang.String id)

Broadcasts messages in this MessageStream to all instances of its downstream operators..

Parameters:

serde - the Serde to use for (de)serializing the message.

id - id the unique id of this operator in this application

Returns:

the broadcast MessageStream