org.apache.samza.operators

Interface MessageStream<M>

Type Parameters:

M - the type of messages in this stream

@InterfaceStability.Unstable
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 StreamGraph.getInputStream(java.lang.String, java.util.function.BiFunction<? super K, ? super V, ? extends M>).

Method Summary

Modifier and Type	Method and Description
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
<K,JM,OM> MessageStream<OM>	join(MessageStream<JM> otherStream, JoinFunction<? extends K,? super M,? super JM,? extends OM> joinFn, java.time.Duration ttl) Joins this MessageStream with another MessageStream using the provided pairwise JoinFunction.
<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> MessageStream<M>	partitionBy(java.util.function.Function<? super M,? extends K> keyExtractor) Re-partitions this MessageStream using keys from the keyExtractor by creating a new intermediate stream on the job.default.system.
<K,V> void	sendTo(OutputStream<K,V,M> outputStream) Allows sending messages in this MessageStream to an OutputStream.
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) 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

filter

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

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

The Function 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 transformed 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

<K,V> void sendTo(OutputStream<K,V,M> outputStream)

Allows sending messages in this MessageStream to an OutputStream.

Type Parameters:

K - the type of key in the outgoing message

V - the type of message in the outgoing message

Parameters:

outputStream - the output stream to send messages to

window

<K,WV> MessageStream<WindowPane<K,WV>> window(Window<M,K,WV> window)

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.

Warning: As of version 0.13.0, messages in windows are kept in memory and will be lost during restarts.

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

Returns:

the transformed MessageStream

join

<K,JM,OM> MessageStream<OM> join(MessageStream<JM> otherStream,
                                 JoinFunction<? extends K,? super M,? super JM,? extends OM> joinFn,
                                 java.time.Duration ttl)

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.

Warning: As of version 0.13.0, messages in joins are kept in memory and will be lost during restarts.

Type Parameters:

K - the type of join key

JM - the type of messages in the other stream

OM - 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

ttl - the ttl for messages in each stream

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> MessageStream<M> partitionBy(java.util.function.Function<? super M,? extends K> keyExtractor)

Re-partitions this MessageStream using keys from the keyExtractor by creating a new intermediate stream on the job.default.system. This intermediate stream is both an output and input to the job.

The key and message Serdes configured for the default system must be able to serialize and deserialize types K and M respectively.

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<JM>, org.apache.samza.operators.functions.JoinFunction<? extends K, ? super M, ? super JM, ? extends OM>, java.time.Duration), 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).

Type Parameters:

K - the type of output message key and partition key

Parameters:

keyExtractor - the Function to extract the output message key and partition key from the input message

Returns:

the repartitioned MessageStream