MessageStream (samza-api 1.9.0-SNAPSHOT API)
JavaScript is disabled on your browser.
Skip navigation links
Overview
Package
Class
Tree
Deprecated
Index
Help
Prev Class
Next Class
Frames
No Frames
All Classes
Summary: 
Nested | 
Field | 
Constr | 
Method
Detail: 
Field | 
Constr | 
Method
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
All Methods Static Methods Instance Methods Abstract Methods Default Methods 
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)
Allows sending messages in this MessageStream to a Table and then propagates this
MessageStream to the next chained operator.
<K,V,U> MessageStream<KV<K,UpdateMessage<U,V>>>
sendTo(Table<KV<K,V>> table,
UpdateOptions updateOptions)
Allows sending update 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)
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
Returns:
this MessageStream
sendTo
<K,V,U> MessageStream<KV<K,UpdateMessage<U,V>>> sendTo(Table<KV<K,V>> table,
UpdateOptions updateOptions)
Allows sending update 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. The value is an UpdateMessage- update and default
value. Defaults are optional and can be used if the Remote Table integration supports inserting a default through
PUT in the event an update fails due to an existing record being absent.
The user also needs to pass UpdateOptions argument which defines whether the update is an update only operation
or a update with default.
Note: The update 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 propagated to next operator.
Type Parameters:
K - the type of key in the table
V - the type of record value in the table
U - the type of update value for the table
Parameters:
table - the table to write update messages to
updateOptions - The update options which defines how the update will be performed
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
Skip navigation links
Overview
Package
Class
Tree
Deprecated
Index
Help
Prev Class
Next Class
Frames
No Frames
All Classes
Summary: 
Nested | 
Field | 
Constr | 
Method
Detail: 
Field | 
Constr | 
Method