public class KStreamImpl<K,V> extends AbstractStream<K,V> implements KStream<K,V>
builder, keySerde, name, sourceNodes, streamsGraphNode, valSerde
Modifier and Type | Method and Description |
---|---|
KStream<K,V>[] |
branch(Predicate<? super K,? super V>... predicates)
Creates an array of
KStream from this stream by branching the records in the original stream based on
the supplied predicates. |
KStream<K,V> |
filter(Predicate<? super K,? super V> predicate)
Create a new
KStream that consists of all records of this stream which satisfy the given predicate. |
KStream<K,V> |
filterNot(Predicate<? super K,? super V> predicate)
Create a new
KStream that consists all records of this stream which do not satisfy the given
predicate. |
<KR,VR> KStream<KR,VR> |
flatMap(KeyValueMapper<? super K,? super V,? extends java.lang.Iterable<? extends KeyValue<? extends KR,? extends VR>>> mapper)
Transform each record of the input stream into zero or more records in the output stream (both key and value type
can be altered arbitrarily).
|
<VR> KStream<K,VR> |
flatMapValues(ValueMapper<? super V,? extends java.lang.Iterable<? extends VR>> mapper)
Create a new
KStream by transforming the value of each record in this stream into zero or more values
with the same key in the new stream. |
<VR> KStream<K,VR> |
flatMapValues(ValueMapperWithKey<? super K,? super V,? extends java.lang.Iterable<? extends VR>> mapper)
Create a new
KStream by transforming the value of each record in this stream into zero or more values
with the same key in the new stream. |
void |
foreach(ForeachAction<? super K,? super V> action)
Perform an action on each record of
KStream . |
<KR> KGroupedStream<KR,V> |
groupBy(KeyValueMapper<? super K,? super V,KR> selector)
Group the records of this
KStream on a new key that is selected using the provided KeyValueMapper
and default serializers and deserializers. |
<KR> KGroupedStream<KR,V> |
groupBy(KeyValueMapper<? super K,? super V,KR> selector,
Grouped<KR,V> grouped)
Group the records of this
KStream on a new key that is selected using the provided KeyValueMapper
and Serde s as specified by Grouped . |
<KR> KGroupedStream<KR,V> |
groupBy(KeyValueMapper<? super K,? super V,KR> selector,
Serialized<KR,V> serialized)
Deprecated.
|
KGroupedStream<K,V> |
groupByKey()
Group the records by their current key into a
KGroupedStream while preserving the original values
and default serializers and deserializers. |
KGroupedStream<K,V> |
groupByKey(Grouped<K,V> grouped)
Group the records by their current key into a
KGroupedStream while preserving the original values
and using the serializers as defined by Grouped . |
KGroupedStream<K,V> |
groupByKey(Serialized<K,V> serialized)
Deprecated.
|
<KG,VG,VR> KStream<K,VR> |
join(GlobalKTable<KG,VG> globalTable,
KeyValueMapper<? super K,? super V,? extends KG> keyMapper,
ValueJoiner<? super V,? super VG,? extends VR> joiner)
Join records of this stream with
GlobalKTable 's records using non-windowed inner equi join. |
<VO,VR> KStream<K,VR> |
join(KStream<K,VO> other,
ValueJoiner<? super V,? super VO,? extends VR> joiner,
JoinWindows windows)
Join records of this stream with another
KStream 's records using windowed inner equi join with default
serializers and deserializers. |
<VO,VR> KStream<K,VR> |
join(KStream<K,VO> otherStream,
ValueJoiner<? super V,? super VO,? extends VR> joiner,
JoinWindows windows,
Joined<K,V,VO> joined)
Join records of this stream with another
KStream 's records using windowed inner equi join using the
Joined instance for configuration of the key serde , this stream's value serde ,
and the other stream's value serde . |
<VO,VR> KStream<K,VR> |
join(KTable<K,VO> other,
ValueJoiner<? super V,? super VO,? extends VR> joiner)
Join records of this stream with
KTable 's records using non-windowed inner equi join with default
serializers and deserializers. |
<VO,VR> KStream<K,VR> |
join(KTable<K,VO> other,
ValueJoiner<? super V,? super VO,? extends VR> joiner,
Joined<K,V,VO> joined)
Join records of this stream with
KTable 's records using non-windowed inner equi join with default
serializers and deserializers. |
<KG,VG,VR> KStream<K,VR> |
leftJoin(GlobalKTable<KG,VG> globalTable,
KeyValueMapper<? super K,? super V,? extends KG> keyMapper,
ValueJoiner<? super V,? super VG,? extends VR> joiner)
Join records of this stream with
GlobalKTable 's records using non-windowed left equi join. |
<VO,VR> KStream<K,VR> |
leftJoin(KStream<K,VO> other,
ValueJoiner<? super V,? super VO,? extends VR> joiner,
JoinWindows windows)
Join records of this stream with another
KStream 's records using windowed left equi join with default
serializers and deserializers. |
<VO,VR> KStream<K,VR> |
leftJoin(KStream<K,VO> other,
ValueJoiner<? super V,? super VO,? extends VR> joiner,
JoinWindows windows,
Joined<K,V,VO> joined)
Join records of this stream with another
KStream 's records using windowed left equi join using the
Joined instance for configuration of the key serde , this stream's value serde ,
and the other stream's value serde . |
<VO,VR> KStream<K,VR> |
leftJoin(KTable<K,VO> other,
ValueJoiner<? super V,? super VO,? extends VR> joiner)
Join records of this stream with
KTable 's records using non-windowed left equi join with default
serializers and deserializers. |
<VO,VR> KStream<K,VR> |
leftJoin(KTable<K,VO> other,
ValueJoiner<? super V,? super VO,? extends VR> joiner,
Joined<K,V,VO> joined)
Join records of this stream with
KTable 's records using non-windowed left equi join with default
serializers and deserializers. |
<KR,VR> KStream<KR,VR> |
map(KeyValueMapper<? super K,? super V,? extends KeyValue<? extends KR,? extends VR>> mapper)
Transform each record of the input stream into a new record in the output stream (both key and value type can be
altered arbitrarily).
|
<VR> KStream<K,VR> |
mapValues(ValueMapper<? super V,? extends VR> mapper)
Transform the value of each input record into a new value (with possible new type) of the output record.
|
<VR> KStream<K,VR> |
mapValues(ValueMapperWithKey<? super K,? super V,? extends VR> mapper)
Transform the value of each input record into a new value (with possible new type) of the output record.
|
KStream<K,V> |
merge(KStream<K,V> stream)
Merge this stream and the given stream into one larger stream.
|
<VO,VR> KStream<K,VR> |
outerJoin(KStream<K,VO> other,
ValueJoiner<? super V,? super VO,? extends VR> joiner,
JoinWindows windows)
Join records of this stream with another
KStream 's records using windowed outer equi join with default
serializers and deserializers. |
<VO,VR> KStream<K,VR> |
outerJoin(KStream<K,VO> other,
ValueJoiner<? super V,? super VO,? extends VR> joiner,
JoinWindows windows,
Joined<K,V,VO> joined)
Join records of this stream with another
KStream 's records using windowed outer equi join using the
Joined instance for configuration of the key serde , this stream's value serde ,
and the other stream's value serde . |
KStream<K,V> |
peek(ForeachAction<? super K,? super V> action)
Perform an action on each record of
KStream . |
void |
print(Printed<K,V> printed)
Print the records of this KStream using the options provided by
Printed
Note that this is mainly for debugging/testing purposes, and it will try to flush on each record print. |
void |
process(ProcessorSupplier<? super K,? super V> processorSupplier,
java.lang.String... stateStoreNames)
Process all records in this stream, one record at a time, by applying a
Processor (provided by the given
ProcessorSupplier ). |
<KR> KStream<KR,V> |
selectKey(KeyValueMapper<? super K,? super V,? extends KR> mapper)
Set a new key (with possibly new type) for each input record.
|
KStream<K,V> |
through(java.lang.String topic)
Materialize this stream to a topic and creates a new
KStream from the topic using default serializers,
deserializers, and producer's DefaultPartitioner . |
KStream<K,V> |
through(java.lang.String topic,
Produced<K,V> produced)
Materialize this stream to a topic and creates a new
KStream from the topic using the
Produced instance for configuration of the key serde , value serde ,
and StreamPartitioner . |
void |
to(java.lang.String topic)
Materialize this stream to a topic using default serializers specified in the config and producer's
DefaultPartitioner . |
void |
to(java.lang.String topic,
Produced<K,V> produced)
Materialize this stream to a topic using the provided
Produced instance. |
void |
to(TopicNameExtractor<K,V> topicExtractor)
Dynamically materialize this stream to topics using default serializers specified in the config and producer's
DefaultPartitioner . |
void |
to(TopicNameExtractor<K,V> topicExtractor,
Produced<K,V> produced)
Dynamically materialize this stream to topics using the provided
Produced instance. |
<KR,VR> KStream<KR,VR> |
transform(TransformerSupplier<? super K,? super V,KeyValue<KR,VR>> transformerSupplier,
java.lang.String... stateStoreNames)
Transform each record of the input stream into zero or more records in the output stream (both key and value type
can be altered arbitrarily).
|
<VR> KStream<K,VR> |
transformValues(ValueTransformerSupplier<? super V,? extends VR> valueTransformerSupplier,
java.lang.String... stateStoreNames)
Transform the value of each input record into a new value (with possible new type) of the output record.
|
<VR> KStream<K,VR> |
transformValues(ValueTransformerWithKeySupplier<? super K,? super V,? extends VR> valueTransformerSupplier,
java.lang.String... stateStoreNames)
Transform the value of each input record into a new value (with possible new type) of the output record.
|
internalTopologyBuilder, keySerde, valueSerde
public KStream<K,V> filter(Predicate<? super K,? super V> predicate)
KStream
KStream
that consists of all records of this stream which satisfy the given predicate.
All records that do not satisfy the predicate are dropped.
This is a stateless record-by-record operation.public KStream<K,V> filterNot(Predicate<? super K,? super V> predicate)
KStream
KStream
that consists all records of this stream which do not satisfy the given
predicate.
All records that do satisfy the predicate are dropped.
This is a stateless record-by-record operation.public <KR> KStream<KR,V> selectKey(KeyValueMapper<? super K,? super V,? extends KR> mapper)
KStream
KeyValueMapper
is applied to each input record and computes a new key for it.
Thus, an input record <K,V>
can be transformed into an output record <K':V>
.
This is a stateless record-by-record operation.
For example, you can use this transformation to set a key for a key-less input record <null,V>
by
extracting a key from the value within your KeyValueMapper
. The example below computes the new key as the
length of the value string.
KStream<Byte[], String> keyLessStream = builder.stream("key-less-topic");
KStream<Integer, String> keyedStream = keyLessStream.selectKey(new KeyValueMapper<Byte[], String, Integer> {
Integer apply(Byte[] key, String value) {
return value.length();
}
});
Setting a new key might result in an internal data redistribution if a key based operator (like an aggregation or
join) is applied to the result KStream
.
selectKey
in interface KStream<K,V>
KR
- the new key type of the result streammapper
- a KeyValueMapper
that computes a new key for each recordKStream
that contains records with new key (possibly of different type) and unmodified valueKStream.map(KeyValueMapper)
,
KStream.flatMap(KeyValueMapper)
,
KStream.mapValues(ValueMapper)
,
KStream.mapValues(ValueMapperWithKey)
,
KStream.flatMapValues(ValueMapper)
,
KStream.flatMapValues(ValueMapperWithKey)
public <KR,VR> KStream<KR,VR> map(KeyValueMapper<? super K,? super V,? extends KeyValue<? extends KR,? extends VR>> mapper)
KStream
KeyValueMapper
is applied to each input record and computes a new output record.
Thus, an input record <K,V>
can be transformed into an output record <K':V'>
.
This is a stateless record-by-record operation (cf. KStream.transform(TransformerSupplier, String...)
for
stateful record transformation).
The example below normalizes the String key to upper-case letters and counts the number of token of the value string.
KStream<String, String> inputStream = builder.stream("topic");
KStream<String, Integer> outputStream = inputStream.map(new KeyValueMapper<String, String, KeyValue<String, Integer>> {
KeyValue<String, Integer> apply(String key, String value) {
return new KeyValue<>(key.toUpperCase(), value.split(" ").length);
}
});
The provided KeyValueMapper
must return a KeyValue
type and must not return null
.
Mapping records might result in an internal data redistribution if a key based operator (like an aggregation or
join) is applied to the result KStream
. (cf. KStream.mapValues(ValueMapper)
)
map
in interface KStream<K,V>
KR
- the key type of the result streamVR
- the value type of the result streammapper
- a KeyValueMapper
that computes a new output recordKStream
that contains records with new key and value (possibly both of different type)KStream.selectKey(KeyValueMapper)
,
KStream.flatMap(KeyValueMapper)
,
KStream.mapValues(ValueMapper)
,
KStream.mapValues(ValueMapperWithKey)
,
KStream.flatMapValues(ValueMapper)
,
KStream.flatMapValues(ValueMapperWithKey)
,
KStream.transform(TransformerSupplier, String...)
,
KStream.transformValues(ValueTransformerSupplier, String...)
,
KStream.transformValues(ValueTransformerWithKeySupplier, String...)
public <VR> KStream<K,VR> mapValues(ValueMapper<? super V,? extends VR> mapper)
KStream
ValueMapper
is applied to each input record value and computes a new value for it.
Thus, an input record <K,V>
can be transformed into an output record <K:V'>
.
This is a stateless record-by-record operation (cf.
KStream.transformValues(ValueTransformerSupplier, String...)
for stateful value transformation).
The example below counts the number of token of the value string.
KStream<String, String> inputStream = builder.stream("topic");
KStream<String, Integer> outputStream = inputStream.mapValues(new ValueMapper<String, Integer> {
Integer apply(String value) {
return value.split(" ").length;
}
});
Setting a new value preserves data co-location with respect to the key.
Thus, no internal data redistribution is required if a key based operator (like an aggregation or join)
is applied to the result KStream
. (cf. KStream.map(KeyValueMapper)
)
mapValues
in interface KStream<K,V>
VR
- the value type of the result streammapper
- a ValueMapper
that computes a new output valueKStream
that contains records with unmodified key and new values (possibly of different type)KStream.selectKey(KeyValueMapper)
,
KStream.map(KeyValueMapper)
,
KStream.flatMap(KeyValueMapper)
,
KStream.flatMapValues(ValueMapper)
,
KStream.flatMapValues(ValueMapperWithKey)
,
KStream.transform(TransformerSupplier, String...)
,
KStream.transformValues(ValueTransformerSupplier, String...)
,
KStream.transformValues(ValueTransformerWithKeySupplier, String...)
public <VR> KStream<K,VR> mapValues(ValueMapperWithKey<? super K,? super V,? extends VR> mapper)
KStream
ValueMapperWithKey
is applied to each input record value and computes a new value for it.
Thus, an input record <K,V>
can be transformed into an output record <K:V'>
.
This is a stateless record-by-record operation (cf.
KStream.transformValues(ValueTransformerWithKeySupplier, String...)
for stateful value transformation).
The example below counts the number of tokens of key and value strings.
KStream<String, String> inputStream = builder.stream("topic");
KStream<String, Integer> outputStream = inputStream.mapValues(new ValueMapperWithKey<String, String, Integer> {
Integer apply(String readOnlyKey, String value) {
return readOnlyKey.split(" ").length + value.split(" ").length;
}
});
Note that the key is read-only and should not be modified, as this can lead to corrupt partitioning.
So, setting a new value preserves data co-location with respect to the key.
Thus, no internal data redistribution is required if a key based operator (like an aggregation or join)
is applied to the result KStream
. (cf. KStream.map(KeyValueMapper)
)
mapValues
in interface KStream<K,V>
VR
- the value type of the result streammapper
- a ValueMapperWithKey
that computes a new output valueKStream
that contains records with unmodified key and new values (possibly of different type)KStream.selectKey(KeyValueMapper)
,
KStream.map(KeyValueMapper)
,
KStream.flatMap(KeyValueMapper)
,
KStream.flatMapValues(ValueMapper)
,
KStream.flatMapValues(ValueMapperWithKey)
,
KStream.transform(TransformerSupplier, String...)
,
KStream.transformValues(ValueTransformerSupplier, String...)
,
KStream.transformValues(ValueTransformerWithKeySupplier, String...)
public void print(Printed<K,V> printed)
KStream
Printed
Note that this is mainly for debugging/testing purposes, and it will try to flush on each record print.
It SHOULD NOT be used for production usage if performance requirements are concerned.public <KR,VR> KStream<KR,VR> flatMap(KeyValueMapper<? super K,? super V,? extends java.lang.Iterable<? extends KeyValue<? extends KR,? extends VR>>> mapper)
KStream
KeyValueMapper
is applied to each input record and computes zero or more output records.
Thus, an input record <K,V>
can be transformed into output records <K':V'>, <K'':V''>, ...
.
This is a stateless record-by-record operation (cf. KStream.transform(TransformerSupplier, String...)
for
stateful record transformation).
The example below splits input records <null:String>
containing sentences as values into their words
and emit a record <word:1>
for each word.
KStream<byte[], String> inputStream = builder.stream("topic");
KStream<String, Integer> outputStream = inputStream.flatMap(new KeyValueMapper<byte[], String, Iterable<KeyValue<String, Integer>>> {
Iterable<KeyValue<String, Integer>> apply(byte[] key, String value) {
String[] tokens = value.split(" ");
List<KeyValue<String, Integer>> result = new ArrayList<>(tokens.length);
for(String token : tokens) {
result.add(new KeyValue<>(token, 1));
}
return result;
}
});
The provided KeyValueMapper
must return an Iterable
(e.g., any Collection
type)
and the return value must not be null
.
Flat-mapping records might result in an internal data redistribution if a key based operator (like an aggregation
or join) is applied to the result KStream
. (cf. KStream.flatMapValues(ValueMapper)
)
flatMap
in interface KStream<K,V>
KR
- the key type of the result streamVR
- the value type of the result streammapper
- a KeyValueMapper
that computes the new output recordsKStream
that contains more or less records with new key and value (possibly of different type)KStream.selectKey(KeyValueMapper)
,
KStream.map(KeyValueMapper)
,
KStream.mapValues(ValueMapper)
,
KStream.mapValues(ValueMapperWithKey)
,
KStream.flatMapValues(ValueMapper)
,
KStream.flatMapValues(ValueMapperWithKey)
,
KStream.transform(TransformerSupplier, String...)
,
KStream.transformValues(ValueTransformerSupplier, String...)
,
KStream.transformValues(ValueTransformerWithKeySupplier, String...)
public <VR> KStream<K,VR> flatMapValues(ValueMapper<? super V,? extends java.lang.Iterable<? extends VR>> mapper)
KStream
KStream
by transforming the value of each record in this stream into zero or more values
with the same key in the new stream.
Transform the value of each input record into zero or more records with the same (unmodified) key in the output
stream (value type can be altered arbitrarily).
The provided ValueMapper
is applied to each input record and computes zero or more output values.
Thus, an input record <K,V>
can be transformed into output records <K:V'>, <K:V''>, ...
.
This is a stateless record-by-record operation (cf. KStream.transformValues(ValueTransformerSupplier, String...)
for stateful value transformation).
The example below splits input records <null:String>
containing sentences as values into their words.
KStream<byte[], String> inputStream = builder.stream("topic");
KStream<byte[], String> outputStream = inputStream.flatMapValues(new ValueMapper<String, Iterable<String>> {
Iterable<String> apply(String value) {
return Arrays.asList(value.split(" "));
}
});
The provided ValueMapper
must return an Iterable
(e.g., any Collection
type)
and the return value must not be null
.
Splitting a record into multiple records with the same key preserves data co-location with respect to the key.
Thus, no internal data redistribution is required if a key based operator (like an aggregation or join)
is applied to the result KStream
. (cf. KStream.flatMap(KeyValueMapper)
)
flatMapValues
in interface KStream<K,V>
VR
- the value type of the result streammapper
- a ValueMapper
the computes the new output valuesKStream
that contains more or less records with unmodified keys and new values of different typeKStream.selectKey(KeyValueMapper)
,
KStream.map(KeyValueMapper)
,
KStream.flatMap(KeyValueMapper)
,
KStream.mapValues(ValueMapper)
,
KStream.mapValues(ValueMapperWithKey)
,
KStream.transform(TransformerSupplier, String...)
,
KStream.transformValues(ValueTransformerSupplier, String...)
,
KStream.transformValues(ValueTransformerWithKeySupplier, String...)
public <VR> KStream<K,VR> flatMapValues(ValueMapperWithKey<? super K,? super V,? extends java.lang.Iterable<? extends VR>> mapper)
KStream
KStream
by transforming the value of each record in this stream into zero or more values
with the same key in the new stream.
Transform the value of each input record into zero or more records with the same (unmodified) key in the output
stream (value type can be altered arbitrarily).
The provided ValueMapperWithKey
is applied to each input record and computes zero or more output values.
Thus, an input record <K,V>
can be transformed into output records <K:V'>, <K:V''>, ...
.
This is a stateless record-by-record operation (cf. KStream.transformValues(ValueTransformerWithKeySupplier, String...)
for stateful value transformation).
The example below splits input records <Integer:String>
, with key=1, containing sentences as values
into their words.
KStream<Integer, String> inputStream = builder.stream("topic");
KStream<Integer, String> outputStream = inputStream.flatMapValues(new ValueMapper<Integer, String, Iterable<String>> {
Iterable<Integer, String> apply(Integer readOnlyKey, String value) {
if(readOnlyKey == 1) {
return Arrays.asList(value.split(" "));
} else {
return Arrays.asList(value);
}
}
});
The provided ValueMapperWithKey
must return an Iterable
(e.g., any Collection
type)
and the return value must not be null
.
Note that the key is read-only and should not be modified, as this can lead to corrupt partitioning.
So, splitting a record into multiple records with the same key preserves data co-location with respect to the key.
Thus, no internal data redistribution is required if a key based operator (like an aggregation or join)
is applied to the result KStream
. (cf. KStream.flatMap(KeyValueMapper)
)
flatMapValues
in interface KStream<K,V>
VR
- the value type of the result streammapper
- a ValueMapperWithKey
the computes the new output valuesKStream
that contains more or less records with unmodified keys and new values of different typeKStream.selectKey(KeyValueMapper)
,
KStream.map(KeyValueMapper)
,
KStream.flatMap(KeyValueMapper)
,
KStream.mapValues(ValueMapper)
,
KStream.mapValues(ValueMapperWithKey)
,
KStream.transform(TransformerSupplier, String...)
,
KStream.transformValues(ValueTransformerSupplier, String...)
,
KStream.transformValues(ValueTransformerWithKeySupplier, String...)
public KStream<K,V>[] branch(Predicate<? super K,? super V>... predicates)
KStream
KStream
from this stream by branching the records in the original stream based on
the supplied predicates.
Each record is evaluated against the supplied predicates, and predicates are evaluated in order.
Each stream in the result array corresponds position-wise (index) to the predicate in the supplied predicates.
The branching happens on first-match: A record in the original stream is assigned to the corresponding result
stream for the first predicate that evaluates to true, and is assigned to this stream only.
A record will be dropped if none of the predicates evaluate to true.
This is a stateless record-by-record operation.public KStream<K,V> merge(KStream<K,V> stream)
KStream
There is no ordering guarantee between records from this KStream
and records from
the provided KStream
in the merged stream.
Relative order is preserved within each input stream though (ie, records within one input
stream are processed in order).
public void foreach(ForeachAction<? super K,? super V> action)
KStream
KStream
.
This is a stateless record-by-record operation (cf. KStream.process(ProcessorSupplier, String...)
).
Note that this is a terminal operation that returns void.foreach
in interface KStream<K,V>
action
- an action to perform on each recordKStream.process(ProcessorSupplier, String...)
public KStream<K,V> peek(ForeachAction<? super K,? super V> action)
KStream
KStream
.
This is a stateless record-by-record operation (cf. KStream.process(ProcessorSupplier, String...)
).
Peek is a non-terminal operation that triggers a side effect (such as logging or statistics collection) and returns an unchanged stream.
Note that since this operation is stateless, it may execute multiple times for a single record in failure cases.
peek
in interface KStream<K,V>
action
- an action to perform on each recordKStream.process(ProcessorSupplier, String...)
public KStream<K,V> through(java.lang.String topic)
KStream
KStream
from the topic using default serializers,
deserializers, and producer's DefaultPartitioner
.
The specified topic should be manually created before it is used (i.e., before the Kafka Streams application is
started).
This is equivalent to calling #to(someTopicName)
and
StreamsBuilder#stream(someTopicName)
.
public KStream<K,V> through(java.lang.String topic, Produced<K,V> produced)
KStream
KStream
from the topic using the
Produced
instance for configuration of the key serde
, value serde
,
and StreamPartitioner
.
The specified topic should be manually created before it is used (i.e., before the Kafka Streams application is
started).
This is equivalent to calling to(someTopic, Produced.with(keySerde, valueSerde)
and StreamsBuilder#stream(someTopicName, Consumed.with(keySerde, valueSerde))
.
public void to(java.lang.String topic)
KStream
DefaultPartitioner
.
The specified topic should be manually created before it is used (i.e., before the Kafka Streams application is
started).public void to(java.lang.String topic, Produced<K,V> produced)
KStream
Produced
instance.
The specified topic should be manually created before it is used (i.e., before the Kafka Streams application is
started).public void to(TopicNameExtractor<K,V> topicExtractor)
KStream
DefaultPartitioner
.
The topic names for each record to send to is dynamically determined based on the TopicNameExtractor
.public void to(TopicNameExtractor<K,V> topicExtractor, Produced<K,V> produced)
KStream
Produced
instance.
The topic names for each record to send to is dynamically determined based on the TopicNameExtractor
.public <KR,VR> KStream<KR,VR> transform(TransformerSupplier<? super K,? super V,KeyValue<KR,VR>> transformerSupplier, java.lang.String... stateStoreNames)
KStream
Transformer
(provided by the given TransformerSupplier
) is applied to each input record and
computes zero or more output records.
Thus, an input record <K,V>
can be transformed into output records <K':V'>, <K'':V''>, ...
.
This is a stateful record-by-record operation (cf. KStream.flatMap(KeyValueMapper)
).
Furthermore, via Punctuator.punctuate(long)
the processing progress can be observed and additional
periodic actions can be performed.
In order to assign a state, the state must be created and registered beforehand:
// create store
StoreBuilder<KeyValueStore<String,String>> keyValueStoreBuilder =
Stores.keyValueStoreBuilder(Stores.persistentKeyValueStore("myTransformState"),
Serdes.String(),
Serdes.String());
// register store
builder.addStateStore(keyValueStoreBuilder);
KStream outputStream = inputStream.transform(new TransformerSupplier() { ... }, "myTransformState");
Within the Transformer
, the state is obtained via the
ProcessorContext
.
To trigger periodic actions via punctuate()
, a schedule must be registered.
The Transformer
must return a KeyValue
type in transform()
and punctuate()
.
new TransformerSupplier() {
Transformer get() {
return new Transformer() {
private ProcessorContext context;
private StateStore state;
void init(ProcessorContext context) {
this.context = context;
this.state = context.getStateStore("myTransformState");
// punctuate each 1000ms; can access this.state
// can emit as many new KeyValue pairs as required via this.context#forward()
context.schedule(Duration.ofSeconds(1), PunctuationType.WALL_CLOCK_TIME, new Punctuator(..));
}
KeyValue transform(K key, V value) {
// can access this.state
// can emit as many new KeyValue pairs as required via this.context#forward()
return new KeyValue(key, value); // can emit a single value via return -- can also be null
}
void close() {
// can access this.state
// can emit as many new KeyValue pairs as required via this.context#forward()
}
}
}
}
Transforming records might result in an internal data redistribution if a key based operator (like an aggregation
or join) is applied to the result KStream
.
(cf. KStream.transformValues(ValueTransformerSupplier, String...)
)
transform
in interface KStream<K,V>
KR
- the key type of the new streamVR
- the value type of the new streamtransformerSupplier
- a instance of TransformerSupplier
that generates a Transformer
stateStoreNames
- the names of the state stores used by the processorKStream
that contains more or less records with new key and value (possibly of different type)KStream.flatMap(KeyValueMapper)
,
KStream.transformValues(ValueTransformerSupplier, String...)
,
KStream.transformValues(ValueTransformerWithKeySupplier, String...)
,
KStream.process(ProcessorSupplier, String...)
public <VR> KStream<K,VR> transformValues(ValueTransformerSupplier<? super V,? extends VR> valueTransformerSupplier, java.lang.String... stateStoreNames)
KStream
ValueTransformer
(provided by the given ValueTransformerSupplier
) is applied to each input
record value and computes a new value for it.
Thus, an input record <K,V>
can be transformed into an output record <K:V'>
.
This is a stateful record-by-record operation (cf. KStream.mapValues(ValueMapper)
).
Furthermore, via Punctuator.punctuate(long)
the processing progress can be observed and additional
periodic actions can be performed.
In order to assign a state, the state must be created and registered beforehand:
// create store
StoreBuilder<KeyValueStore<String,String>> keyValueStoreBuilder =
Stores.keyValueStoreBuilder(Stores.persistentKeyValueStore("myValueTransformState"),
Serdes.String(),
Serdes.String());
// register store
builder.addStateStore(keyValueStoreBuilder);
KStream outputStream = inputStream.transformValues(new ValueTransformerSupplier() { ... }, "myValueTransformState");
Within the ValueTransformer
, the state is obtained via the
ProcessorContext
.
To trigger periodic actions via punctuate()
, a schedule must be
registered.
In contrast to transform()
, no additional KeyValue
pairs should be emitted via ProcessorContext.forward()
.
new ValueTransformerSupplier() {
ValueTransformer get() {
return new ValueTransformer() {
private StateStore state;
void init(ProcessorContext context) {
this.state = context.getStateStore("myValueTransformState");
context.schedule(Duration.ofSeconds(1), PunctuationType.WALL_CLOCK_TIME, new Punctuator(..)); // punctuate each 1000ms, can access this.state
}
NewValueType transform(V value) {
// can access this.state
return new NewValueType(); // or null
}
void close() {
// can access this.state
}
}
}
}
Setting a new value preserves data co-location with respect to the key.
Thus, no internal data redistribution is required if a key based operator (like an aggregation or join)
is applied to the result KStream
. (cf. KStream.transform(TransformerSupplier, String...)
)
transformValues
in interface KStream<K,V>
VR
- the value type of the result streamvalueTransformerSupplier
- a instance of ValueTransformerSupplier
that generates a
ValueTransformer
stateStoreNames
- the names of the state stores used by the processorKStream
that contains records with unmodified key and new values (possibly of different type)KStream.mapValues(ValueMapper)
,
KStream.mapValues(ValueMapperWithKey)
,
KStream.transform(TransformerSupplier, String...)
public <VR> KStream<K,VR> transformValues(ValueTransformerWithKeySupplier<? super K,? super V,? extends VR> valueTransformerSupplier, java.lang.String... stateStoreNames)
KStream
ValueTransformerWithKey
(provided by the given ValueTransformerWithKeySupplier
) is applied to each input
record value and computes a new value for it.
Thus, an input record <K,V>
can be transformed into an output record <K:V'>
.
This is a stateful record-by-record operation (cf. KStream.mapValues(ValueMapperWithKey)
).
Furthermore, via Punctuator.punctuate(long)
the processing progress can be observed and additional
periodic actions can be performed.
In order to assign a state, the state must be created and registered beforehand:
// create store
StoreBuilder<KeyValueStore<String,String>> keyValueStoreBuilder =
Stores.keyValueStoreBuilder(Stores.persistentKeyValueStore("myValueTransformState"),
Serdes.String(),
Serdes.String());
// register store
builder.addStateStore(keyValueStoreBuilder);
KStream outputStream = inputStream.transformValues(new ValueTransformerWithKeySupplier() { ... }, "myValueTransformState");
Within the ValueTransformerWithKey
, the state is obtained via the
ProcessorContext
.
To trigger periodic actions via punctuate()
,
a schedule must be registered.
In contrast to transform()
, no additional KeyValue
pairs should be emitted via ProcessorContext.forward()
.
new ValueTransformerWithKeySupplier() {
ValueTransformerWithKey get() {
return new ValueTransformerWithKey() {
private StateStore state;
void init(ProcessorContext context) {
this.state = context.getStateStore("myValueTransformState");
context.schedule(Duration.ofSeconds(1), PunctuationType.WALL_CLOCK_TIME, new Punctuator(..)); // punctuate each 1000ms, can access this.state
}
NewValueType transform(K readOnlyKey, V value) {
// can access this.state and use read-only key
return new NewValueType(readOnlyKey); // or null
}
void close() {
// can access this.state
}
}
}
}
Note that the key is read-only and should not be modified, as this can lead to corrupt partitioning.
So, setting a new value preserves data co-location with respect to the key.
Thus, no internal data redistribution is required if a key based operator (like an aggregation or join)
is applied to the result KStream
. (cf. KStream.transform(TransformerSupplier, String...)
)
transformValues
in interface KStream<K,V>
VR
- the value type of the result streamvalueTransformerSupplier
- a instance of ValueTransformerWithKeySupplier
that generates a
ValueTransformerWithKey
stateStoreNames
- the names of the state stores used by the processorKStream
that contains records with unmodified key and new values (possibly of different type)KStream.mapValues(ValueMapper)
,
KStream.mapValues(ValueMapperWithKey)
,
KStream.transform(TransformerSupplier, String...)
public void process(ProcessorSupplier<? super K,? super V> processorSupplier, java.lang.String... stateStoreNames)
KStream
Processor
(provided by the given
ProcessorSupplier
).
This is a stateful record-by-record operation (cf. KStream.foreach(ForeachAction)
).
Furthermore, via Punctuator.punctuate(long)
the processing progress can be observed and additional
periodic actions can be performed.
Note that this is a terminal operation that returns void.
In order to assign a state, the state must be created and registered beforehand:
// create store
StoreBuilder<KeyValueStore<String,String>> keyValueStoreBuilder =
Stores.keyValueStoreBuilder(Stores.persistentKeyValueStore("myProcessorState"),
Serdes.String(),
Serdes.String());
// register store
builder.addStateStore(keyValueStoreBuilder);
inputStream.process(new ProcessorSupplier() { ... }, "myProcessorState");
Within the Processor
, the state is obtained via the
ProcessorContext
.
To trigger periodic actions via punctuate()
,
a schedule must be registered.
new ProcessorSupplier() {
Processor get() {
return new Processor() {
private StateStore state;
void init(ProcessorContext context) {
this.state = context.getStateStore("myProcessorState");
context.schedule(Duration.ofSeconds(1), PunctuationType.WALL_CLOCK_TIME, new Punctuator(..)); // punctuate each 1000ms, can access this.state
}
void process(K key, V value) {
// can access this.state
}
void close() {
// can access this.state
}
}
}
}
process
in interface KStream<K,V>
processorSupplier
- a instance of ProcessorSupplier
that generates a Processor
stateStoreNames
- the names of the state store used by the processorKStream.foreach(ForeachAction)
,
KStream.transform(TransformerSupplier, String...)
public <VO,VR> KStream<K,VR> join(KStream<K,VO> other, ValueJoiner<? super V,? super VO,? extends VR> joiner, JoinWindows windows)
KStream
KStream
's records using windowed inner equi join with default
serializers and deserializers.
The join is computed on the records' key with join attribute thisKStream.key == otherKStream.key
.
Furthermore, two records are only joined if their timestamps are close to each other as defined by the given
JoinWindows
, i.e., the window defines an additional join predicate on the record timestamps.
For each pair of records meeting both join predicates the provided ValueJoiner
will be called to compute
a value (with arbitrary type) for the result record.
The key of the result record is the same as for both joining input records.
If an input record key or value is null
the record will not be included in the join operation and thus no
output record will be added to the resulting KStream
.
Example (assuming all input records belong to the correct windows):
this | other | result |
---|---|---|
<K1:A> | ||
<K2:B> | <K2:b> | <K2:ValueJoiner(B,b)> |
<K3:c> |
KStream.through(String)
(for one input stream) before doing the
join, using a pre-created topic with the "correct" number of partitions.
Furthermore, both input streams need to be co-partitioned on the join key (i.e., use the same partitioner).
If this requirement is not met, Kafka Streams will automatically repartition the data, i.e., it will create an
internal repartitioning topic in Kafka and write and re-read the data via this topic before the actual join.
The repartitioning topic will be named "${applicationId}-<name>-repartition", where "applicationId" is
user-specified in StreamsConfig
via parameter
APPLICATION_ID_CONFIG
, "<name>" is an internally generated name, and
"-repartition" is a fixed suffix.
Repartitioning can happen for one or both of the joining KStream
s.
For this case, all data of the stream will be redistributed through the repartitioning topic by writing all
records to it, and rereading all records from it, such that the join input KStream
is partitioned
correctly on its key.
Both of the joining KStream
s will be materialized in local state stores with auto-generated store names.
For failure and recovery each store will be backed by an internal changelog topic that will be created in Kafka.
The changelog topic will be named "${applicationId}-storeName-changelog", where "applicationId" is user-specified
in StreamsConfig
via parameter
APPLICATION_ID_CONFIG
, "storeName" is an
internally generated name, and "-changelog" is a fixed suffix.
You can retrieve all generated internal topic names via Topology.describe()
.
join
in interface KStream<K,V>
VO
- the value type of the other streamVR
- the value type of the result streamother
- the KStream
to be joined with this streamjoiner
- a ValueJoiner
that computes the join result for a pair of matching recordswindows
- the specification of the JoinWindows
KStream
that contains join-records for each key and values computed by the given
ValueJoiner
, one for each matched record-pair with the same key and within the joining window intervalsKStream.leftJoin(KStream, ValueJoiner, JoinWindows)
,
KStream.outerJoin(KStream, ValueJoiner, JoinWindows)
public <VO,VR> KStream<K,VR> join(KStream<K,VO> otherStream, ValueJoiner<? super V,? super VO,? extends VR> joiner, JoinWindows windows, Joined<K,V,VO> joined)
KStream
KStream
's records using windowed inner equi join using the
Joined
instance for configuration of the key serde
, this stream's value serde
,
and the other stream's value serde
.
The join is computed on the records' key with join attribute thisKStream.key == otherKStream.key
.
Furthermore, two records are only joined if their timestamps are close to each other as defined by the given
JoinWindows
, i.e., the window defines an additional join predicate on the record timestamps.
For each pair of records meeting both join predicates the provided ValueJoiner
will be called to compute
a value (with arbitrary type) for the result record.
The key of the result record is the same as for both joining input records.
If an input record key or value is null
the record will not be included in the join operation and thus no
output record will be added to the resulting KStream
.
Example (assuming all input records belong to the correct windows):
this | other | result |
---|---|---|
<K1:A> | ||
<K2:B> | <K2:b> | <K2:ValueJoiner(B,b)> |
<K3:c> |
KStream.through(String)
(for one input stream) before doing the
join, using a pre-created topic with the "correct" number of partitions.
Furthermore, both input streams need to be co-partitioned on the join key (i.e., use the same partitioner).
If this requirement is not met, Kafka Streams will automatically repartition the data, i.e., it will create an
internal repartitioning topic in Kafka and write and re-read the data via this topic before the actual join.
The repartitioning topic will be named "${applicationId}-<name>-repartition", where "applicationId" is
user-specified in StreamsConfig
via parameter
APPLICATION_ID_CONFIG
, "<name>" is an internally generated name, and
"-repartition" is a fixed suffix.
Repartitioning can happen for one or both of the joining KStream
s.
For this case, all data of the stream will be redistributed through the repartitioning topic by writing all
records to it, and rereading all records from it, such that the join input KStream
is partitioned
correctly on its key.
Both of the joining KStream
s will be materialized in local state stores with auto-generated store names.
For failure and recovery each store will be backed by an internal changelog topic that will be created in Kafka.
The changelog topic will be named "${applicationId}-storeName-changelog", where "applicationId" is user-specified
in StreamsConfig
via parameter
APPLICATION_ID_CONFIG
, "storeName" is an
internally generated name, and "-changelog" is a fixed suffix.
You can retrieve all generated internal topic names via Topology.describe()
.
join
in interface KStream<K,V>
VO
- the value type of the other streamVR
- the value type of the result streamotherStream
- the KStream
to be joined with this streamjoiner
- a ValueJoiner
that computes the join result for a pair of matching recordswindows
- the specification of the JoinWindows
joined
- a Joined
instance that defines the serdes to
be used to serialize/deserialize inputs and outputs of the joined streamsKStream
that contains join-records for each key and values computed by the given
ValueJoiner
, one for each matched record-pair with the same key and within the joining window intervalsKStream.leftJoin(KStream, ValueJoiner, JoinWindows, Joined)
,
KStream.outerJoin(KStream, ValueJoiner, JoinWindows, Joined)
public <VO,VR> KStream<K,VR> outerJoin(KStream<K,VO> other, ValueJoiner<? super V,? super VO,? extends VR> joiner, JoinWindows windows)
KStream
KStream
's records using windowed outer equi join with default
serializers and deserializers.
In contrast to inner-join
or
left-join
, all records from both streams will produce at
least one output record (cf. below).
The join is computed on the records' key with join attribute thisKStream.key == otherKStream.key
.
Furthermore, two records are only joined if their timestamps are close to each other as defined by the given
JoinWindows
, i.e., the window defines an additional join predicate on the record timestamps.
For each pair of records meeting both join predicates the provided ValueJoiner
will be called to compute
a value (with arbitrary type) for the result record.
The key of the result record is the same as for both joining input records.
Furthermore, for each input record of both KStream
s that does not satisfy the join predicate the provided
ValueJoiner
will be called with a null
value for the this/other stream, respectively.
If an input record key or value is null
the record will not be included in the join operation and thus no
output record will be added to the resulting KStream
.
Example (assuming all input records belong to the correct windows):
this | other | result |
---|---|---|
<K1:A> | <K1:ValueJoiner(A,null)> | |
<K2:B> | <K2:b> | <K2:ValueJoiner(null,b)> <K2:ValueJoiner(B,b)> |
<K3:c> | <K3:ValueJoiner(null,c)> |
KStream.through(String)
(for one input stream) before doing the
join, using a pre-created topic with the "correct" number of partitions.
Furthermore, both input streams need to be co-partitioned on the join key (i.e., use the same partitioner).
If this requirement is not met, Kafka Streams will automatically repartition the data, i.e., it will create an
internal repartitioning topic in Kafka and write and re-read the data via this topic before the actual join.
The repartitioning topic will be named "${applicationId}-<name>-repartition", where "applicationId" is
user-specified in StreamsConfig
via parameter
APPLICATION_ID_CONFIG
, "<name>" is an internally generated name, and
"-repartition" is a fixed suffix.
Repartitioning can happen for one or both of the joining KStream
s.
For this case, all data of the stream will be redistributed through the repartitioning topic by writing all
records to it, and rereading all records from it, such that the join input KStream
is partitioned
correctly on its key.
Both of the joining KStream
s will be materialized in local state stores with auto-generated store names.
For failure and recovery each store will be backed by an internal changelog topic that will be created in Kafka.
The changelog topic will be named "${applicationId}-storeName-changelog", where "applicationId" is user-specified
in StreamsConfig
via parameter APPLICATION_ID_CONFIG
,
"storeName" is an internally generated name, and "-changelog" is a fixed suffix.
You can retrieve all generated internal topic names via Topology.describe()
.
outerJoin
in interface KStream<K,V>
VO
- the value type of the other streamVR
- the value type of the result streamother
- the KStream
to be joined with this streamjoiner
- a ValueJoiner
that computes the join result for a pair of matching recordswindows
- the specification of the JoinWindows
KStream
that contains join-records for each key and values computed by the given
ValueJoiner
, one for each matched record-pair with the same key plus one for each non-matching record of
both KStream
and within the joining window intervalsKStream.join(KStream, ValueJoiner, JoinWindows)
,
KStream.leftJoin(KStream, ValueJoiner, JoinWindows)
public <VO,VR> KStream<K,VR> outerJoin(KStream<K,VO> other, ValueJoiner<? super V,? super VO,? extends VR> joiner, JoinWindows windows, Joined<K,V,VO> joined)
KStream
KStream
's records using windowed outer equi join using the
Joined
instance for configuration of the key serde
, this stream's value serde
,
and the other stream's value serde
.
In contrast to inner-join
or
left-join
, all records from both streams will produce at
least one output record (cf. below).
The join is computed on the records' key with join attribute thisKStream.key == otherKStream.key
.
Furthermore, two records are only joined if their timestamps are close to each other as defined by the given
JoinWindows
, i.e., the window defines an additional join predicate on the record timestamps.
For each pair of records meeting both join predicates the provided ValueJoiner
will be called to compute
a value (with arbitrary type) for the result record.
The key of the result record is the same as for both joining input records.
Furthermore, for each input record of both KStream
s that does not satisfy the join predicate the provided
ValueJoiner
will be called with a null
value for this/other stream, respectively.
If an input record key or value is null
the record will not be included in the join operation and thus no
output record will be added to the resulting KStream
.
Example (assuming all input records belong to the correct windows):
this | other | result |
---|---|---|
<K1:A> | <K1:ValueJoiner(A,null)> | |
<K2:B> | <K2:b> | <K2:ValueJoiner(null,b)> <K2:ValueJoiner(B,b)> |
<K3:c> | <K3:ValueJoiner(null,c)> |
KStream.through(String)
(for one input stream) before doing the
join, using a pre-created topic with the "correct" number of partitions.
Furthermore, both input streams need to be co-partitioned on the join key (i.e., use the same partitioner).
If this requirement is not met, Kafka Streams will automatically repartition the data, i.e., it will create an
internal repartitioning topic in Kafka and write and re-read the data via this topic before the actual join.
The repartitioning topic will be named "${applicationId}-<name>-repartition", where "applicationId" is
user-specified in StreamsConfig
via parameter
APPLICATION_ID_CONFIG
, "<name>" is an internally generated name, and
"-repartition" is a fixed suffix.
Repartitioning can happen for one or both of the joining KStream
s.
For this case, all data of the stream will be redistributed through the repartitioning topic by writing all
records to it, and rereading all records from it, such that the join input KStream
is partitioned
correctly on its key.
Both of the joining KStream
s will be materialized in local state stores with auto-generated store names.
For failure and recovery each store will be backed by an internal changelog topic that will be created in Kafka.
The changelog topic will be named "${applicationId}-storeName-changelog", where "applicationId" is user-specified
in StreamsConfig
via parameter APPLICATION_ID_CONFIG
,
"storeName" is an internally generated name, and "-changelog" is a fixed suffix.
You can retrieve all generated internal topic names via Topology.describe()
.
outerJoin
in interface KStream<K,V>
VO
- the value type of the other streamVR
- the value type of the result streamother
- the KStream
to be joined with this streamjoiner
- a ValueJoiner
that computes the join result for a pair of matching recordswindows
- the specification of the JoinWindows
KStream
that contains join-records for each key and values computed by the given
ValueJoiner
, one for each matched record-pair with the same key plus one for each non-matching record of
both KStream
and within the joining window intervalsKStream.join(KStream, ValueJoiner, JoinWindows, Joined)
,
KStream.leftJoin(KStream, ValueJoiner, JoinWindows, Joined)
public <VO,VR> KStream<K,VR> leftJoin(KStream<K,VO> other, ValueJoiner<? super V,? super VO,? extends VR> joiner, JoinWindows windows)
KStream
KStream
's records using windowed left equi join with default
serializers and deserializers.
In contrast to inner-join
, all records from this stream will
produce at least one output record (cf. below).
The join is computed on the records' key with join attribute thisKStream.key == otherKStream.key
.
Furthermore, two records are only joined if their timestamps are close to each other as defined by the given
JoinWindows
, i.e., the window defines an additional join predicate on the record timestamps.
For each pair of records meeting both join predicates the provided ValueJoiner
will be called to compute
a value (with arbitrary type) for the result record.
The key of the result record is the same as for both joining input records.
Furthermore, for each input record of this KStream
that does not satisfy the join predicate the provided
ValueJoiner
will be called with a null
value for the other stream.
If an input record key or value is null
the record will not be included in the join operation and thus no
output record will be added to the resulting KStream
.
Example (assuming all input records belong to the correct windows):
this | other | result |
---|---|---|
<K1:A> | <K1:ValueJoiner(A,null)> | |
<K2:B> | <K2:b> | <K2:ValueJoiner(B,b)> |
<K3:c> |
KStream.through(String)
(for one input stream) before doing the
join, using a pre-created topic with the "correct" number of partitions.
Furthermore, both input streams need to be co-partitioned on the join key (i.e., use the same partitioner).
If this requirement is not met, Kafka Streams will automatically repartition the data, i.e., it will create an
internal repartitioning topic in Kafka and write and re-read the data via this topic before the actual join.
The repartitioning topic will be named "${applicationId}-<name>-repartition", where "applicationId" is
user-specified in StreamsConfig
via parameter
APPLICATION_ID_CONFIG
, "<name>" is an internally generated name, and
"-repartition" is a fixed suffix.
Repartitioning can happen for one or both of the joining KStream
s.
For this case, all data of the stream will be redistributed through the repartitioning topic by writing all
records to it, and rereading all records from it, such that the join input KStream
is partitioned
correctly on its key.
Both of the joining KStream
s will be materialized in local state stores with auto-generated store names.
For failure and recovery each store will be backed by an internal changelog topic that will be created in Kafka.
The changelog topic will be named "${applicationId}-storeName-changelog", where "applicationId" is user-specified
in StreamsConfig
via parameter APPLICATION_ID_CONFIG
,
"storeName" is an internally generated name, and "-changelog" is a fixed suffix.
You can retrieve all generated internal topic names via Topology.describe()
.
leftJoin
in interface KStream<K,V>
VO
- the value type of the other streamVR
- the value type of the result streamother
- the KStream
to be joined with this streamjoiner
- a ValueJoiner
that computes the join result for a pair of matching recordswindows
- the specification of the JoinWindows
KStream
that contains join-records for each key and values computed by the given
ValueJoiner
, one for each matched record-pair with the same key plus one for each non-matching record of
this KStream
and within the joining window intervalsKStream.join(KStream, ValueJoiner, JoinWindows)
,
KStream.outerJoin(KStream, ValueJoiner, JoinWindows)
public <VO,VR> KStream<K,VR> leftJoin(KStream<K,VO> other, ValueJoiner<? super V,? super VO,? extends VR> joiner, JoinWindows windows, Joined<K,V,VO> joined)
KStream
KStream
's records using windowed left equi join using the
Joined
instance for configuration of the key serde
, this stream's value serde
,
and the other stream's value serde
.
In contrast to inner-join
, all records from this stream will
produce at least one output record (cf. below).
The join is computed on the records' key with join attribute thisKStream.key == otherKStream.key
.
Furthermore, two records are only joined if their timestamps are close to each other as defined by the given
JoinWindows
, i.e., the window defines an additional join predicate on the record timestamps.
For each pair of records meeting both join predicates the provided ValueJoiner
will be called to compute
a value (with arbitrary type) for the result record.
The key of the result record is the same as for both joining input records.
Furthermore, for each input record of this KStream
that does not satisfy the join predicate the provided
ValueJoiner
will be called with a null
value for the other stream.
If an input record key or value is null
the record will not be included in the join operation and thus no
output record will be added to the resulting KStream
.
Example (assuming all input records belong to the correct windows):
this | other | result |
---|---|---|
<K1:A> | <K1:ValueJoiner(A,null)> | |
<K2:B> | <K2:b> | <K2:ValueJoiner(B,b)> |
<K3:c> |
KStream.through(String)
(for one input stream) before doing the
join, using a pre-created topic with the "correct" number of partitions.
Furthermore, both input streams need to be co-partitioned on the join key (i.e., use the same partitioner).
If this requirement is not met, Kafka Streams will automatically repartition the data, i.e., it will create an
internal repartitioning topic in Kafka and write and re-read the data via this topic before the actual join.
The repartitioning topic will be named "${applicationId}-<name>-repartition", where "applicationId" is
user-specified in StreamsConfig
via parameter
APPLICATION_ID_CONFIG
, "<name>" is an internally generated name, and
"-repartition" is a fixed suffix.
Repartitioning can happen for one or both of the joining KStream
s.
For this case, all data of the stream will be redistributed through the repartitioning topic by writing all
records to it, and rereading all records from it, such that the join input KStream
is partitioned
correctly on its key.
Both of the joining KStream
s will be materialized in local state stores with auto-generated store names.
For failure and recovery each store will be backed by an internal changelog topic that will be created in Kafka.
The changelog topic will be named "${applicationId}-storeName-changelog", where "applicationId" is user-specified
in StreamsConfig
via parameter APPLICATION_ID_CONFIG
,
"storeName" is an internally generated name, and "-changelog" is a fixed suffix.
You can retrieve all generated internal topic names via Topology.describe()
.
leftJoin
in interface KStream<K,V>
VO
- the value type of the other streamVR
- the value type of the result streamother
- the KStream
to be joined with this streamjoiner
- a ValueJoiner
that computes the join result for a pair of matching recordswindows
- the specification of the JoinWindows
joined
- a Joined
instance that defines the serdes to
be used to serialize/deserialize inputs and outputs of the joined streamsKStream
that contains join-records for each key and values computed by the given
ValueJoiner
, one for each matched record-pair with the same key plus one for each non-matching record of
this KStream
and within the joining window intervalsKStream.join(KStream, ValueJoiner, JoinWindows, Joined)
,
KStream.outerJoin(KStream, ValueJoiner, JoinWindows, Joined)
public <VO,VR> KStream<K,VR> join(KTable<K,VO> other, ValueJoiner<? super V,? super VO,? extends VR> joiner)
KStream
KTable
's records using non-windowed inner equi join with default
serializers and deserializers.
The join is a primary key table lookup join with join attribute stream.key == table.key
.
"Table lookup join" means, that results are only computed if KStream
records are processed.
This is done by performing a lookup for matching records in the current (i.e., processing time) internal
KTable
state.
In contrast, processing KTable
input records will only update the internal KTable
state and
will not produce any result records.
For each KStream
record that finds a corresponding record in KTable
the provided
ValueJoiner
will be called to compute a value (with arbitrary type) for the result record.
The key of the result record is the same as for both joining input records.
If an KStream
input record key or value is null
the record will not be included in the join
operation and thus no output record will be added to the resulting KStream
.
Example:
KStream | KTable | state | result |
---|---|---|---|
<K1:A> | |||
<K1:b> | <K1:b> | ||
<K1:C> | <K1:b> | <K1:ValueJoiner(C,b)> |
KStream.through(String)
for this KStream
before doing
the join, using a pre-created topic with the same number of partitions as the given KTable
.
Furthermore, both input streams need to be co-partitioned on the join key (i.e., use the same partitioner);
cf. KStream.join(GlobalKTable, KeyValueMapper, ValueJoiner)
.
If this requirement is not met, Kafka Streams will automatically repartition the data, i.e., it will create an
internal repartitioning topic in Kafka and write and re-read the data via this topic before the actual join.
The repartitioning topic will be named "${applicationId}-<name>-repartition", where "applicationId" is
user-specified in StreamsConfig
via parameter
APPLICATION_ID_CONFIG
, "<name>" is an internally generated name, and
"-repartition" is a fixed suffix.
You can retrieve all generated internal topic names via Topology.describe()
.
Repartitioning can happen only for this KStream
but not for the provided KTable
.
For this case, all data of the stream will be redistributed through the repartitioning topic by writing all
records to it, and rereading all records from it, such that the join input KStream
is partitioned
correctly on its key.
join
in interface KStream<K,V>
VO
- the value type of the tableVR
- the value type of the result streamother
- the KTable
to be joined with this streamjoiner
- a ValueJoiner
that computes the join result for a pair of matching recordsKStream
that contains join-records for each key and values computed by the given
ValueJoiner
, one for each matched record-pair with the same keyKStream.leftJoin(KTable, ValueJoiner)
,
KStream.join(GlobalKTable, KeyValueMapper, ValueJoiner)
public <VO,VR> KStream<K,VR> join(KTable<K,VO> other, ValueJoiner<? super V,? super VO,? extends VR> joiner, Joined<K,V,VO> joined)
KStream
KTable
's records using non-windowed inner equi join with default
serializers and deserializers.
The join is a primary key table lookup join with join attribute stream.key == table.key
.
"Table lookup join" means, that results are only computed if KStream
records are processed.
This is done by performing a lookup for matching records in the current (i.e., processing time) internal
KTable
state.
In contrast, processing KTable
input records will only update the internal KTable
state and
will not produce any result records.
For each KStream
record that finds a corresponding record in KTable
the provided
ValueJoiner
will be called to compute a value (with arbitrary type) for the result record.
The key of the result record is the same as for both joining input records.
If an KStream
input record key or value is null
the record will not be included in the join
operation and thus no output record will be added to the resulting KStream
.
Example:
KStream | KTable | state | result |
---|---|---|---|
<K1:A> | |||
<K1:b> | <K1:b> | ||
<K1:C> | <K1:b> | <K1:ValueJoiner(C,b)> |
KStream.through(String)
for this KStream
before doing
the join, using a pre-created topic with the same number of partitions as the given KTable
.
Furthermore, both input streams need to be co-partitioned on the join key (i.e., use the same partitioner);
cf. KStream.join(GlobalKTable, KeyValueMapper, ValueJoiner)
.
If this requirement is not met, Kafka Streams will automatically repartition the data, i.e., it will create an
internal repartitioning topic in Kafka and write and re-read the data via this topic before the actual join.
The repartitioning topic will be named "${applicationId}-<name>-repartition", where "applicationId" is
user-specified in StreamsConfig
via parameter
APPLICATION_ID_CONFIG
, "<name>" is an internally generated name, and
"-repartition" is a fixed suffix.
You can retrieve all generated internal topic names via Topology.describe()
.
Repartitioning can happen only for this KStream
but not for the provided KTable
.
For this case, all data of the stream will be redistributed through the repartitioning topic by writing all
records to it, and rereading all records from it, such that the join input KStream
is partitioned
correctly on its key.
join
in interface KStream<K,V>
VO
- the value type of the tableVR
- the value type of the result streamother
- the KTable
to be joined with this streamjoiner
- a ValueJoiner
that computes the join result for a pair of matching recordsjoined
- a Joined
instance that defines the serdes to
be used to serialize/deserialize inputs of the joined streamsKStream
that contains join-records for each key and values computed by the given
ValueJoiner
, one for each matched record-pair with the same keyKStream.leftJoin(KTable, ValueJoiner, Joined)
,
KStream.join(GlobalKTable, KeyValueMapper, ValueJoiner)
public <VO,VR> KStream<K,VR> leftJoin(KTable<K,VO> other, ValueJoiner<? super V,? super VO,? extends VR> joiner)
KStream
KTable
's records using non-windowed left equi join with default
serializers and deserializers.
In contrast to inner-join
, all records from this stream will produce an
output record (cf. below).
The join is a primary key table lookup join with join attribute stream.key == table.key
.
"Table lookup join" means, that results are only computed if KStream
records are processed.
This is done by performing a lookup for matching records in the current (i.e., processing time) internal
KTable
state.
In contrast, processing KTable
input records will only update the internal KTable
state and
will not produce any result records.
For each KStream
record whether or not it finds a corresponding record in KTable
the provided
ValueJoiner
will be called to compute a value (with arbitrary type) for the result record.
If no KTable
record was found during lookup, a null
value will be provided to ValueJoiner
.
The key of the result record is the same as for both joining input records.
If an KStream
input record key or value is null
the record will not be included in the join
operation and thus no output record will be added to the resulting KStream
.
Example:
KStream | KTable | state | result |
---|---|---|---|
<K1:A> | <K1:ValueJoiner(A,null)> | ||
<K1:b> | <K1:b> | ||
<K1:C> | <K1:b> | <K1:ValueJoiner(C,b)> |
KStream.through(String)
for this KStream
before doing
the join, using a pre-created topic with the same number of partitions as the given KTable
.
Furthermore, both input streams need to be co-partitioned on the join key (i.e., use the same partitioner);
cf. KStream.join(GlobalKTable, KeyValueMapper, ValueJoiner)
.
If this requirement is not met, Kafka Streams will automatically repartition the data, i.e., it will create an
internal repartitioning topic in Kafka and write and re-read the data via this topic before the actual join.
The repartitioning topic will be named "${applicationId}-<name>-repartition", where "applicationId" is
user-specified in StreamsConfig
via parameter
APPLICATION_ID_CONFIG
, "<name>" is an internally generated name, and
"-repartition" is a fixed suffix.
You can retrieve all generated internal topic names via Topology.describe()
.
Repartitioning can happen only for this KStream
but not for the provided KTable
.
For this case, all data of the stream will be redistributed through the repartitioning topic by writing all
records to it, and rereading all records from it, such that the join input KStream
is partitioned
correctly on its key.
leftJoin
in interface KStream<K,V>
VO
- the value type of the tableVR
- the value type of the result streamother
- the KTable
to be joined with this streamjoiner
- a ValueJoiner
that computes the join result for a pair of matching recordsKStream
that contains join-records for each key and values computed by the given
ValueJoiner
, one output for each input KStream
recordKStream.join(KTable, ValueJoiner)
,
KStream.leftJoin(GlobalKTable, KeyValueMapper, ValueJoiner)
public <VO,VR> KStream<K,VR> leftJoin(KTable<K,VO> other, ValueJoiner<? super V,? super VO,? extends VR> joiner, Joined<K,V,VO> joined)
KStream
KTable
's records using non-windowed left equi join with default
serializers and deserializers.
In contrast to inner-join
, all records from this stream will produce an
output record (cf. below).
The join is a primary key table lookup join with join attribute stream.key == table.key
.
"Table lookup join" means, that results are only computed if KStream
records are processed.
This is done by performing a lookup for matching records in the current (i.e., processing time) internal
KTable
state.
In contrast, processing KTable
input records will only update the internal KTable
state and
will not produce any result records.
For each KStream
record whether or not it finds a corresponding record in KTable
the provided
ValueJoiner
will be called to compute a value (with arbitrary type) for the result record.
If no KTable
record was found during lookup, a null
value will be provided to ValueJoiner
.
The key of the result record is the same as for both joining input records.
If an KStream
input record key or value is null
the record will not be included in the join
operation and thus no output record will be added to the resulting KStream
.
Example:
KStream | KTable | state | result |
---|---|---|---|
<K1:A> | <K1:ValueJoiner(A,null)> | ||
<K1:b> | <K1:b> | ||
<K1:C> | <K1:b> | <K1:ValueJoiner(C,b)> |
KStream.through(String)
for this KStream
before doing
the join, using a pre-created topic with the same number of partitions as the given KTable
.
Furthermore, both input streams need to be co-partitioned on the join key (i.e., use the same partitioner);
cf. KStream.join(GlobalKTable, KeyValueMapper, ValueJoiner)
.
If this requirement is not met, Kafka Streams will automatically repartition the data, i.e., it will create an
internal repartitioning topic in Kafka and write and re-read the data via this topic before the actual join.
The repartitioning topic will be named "${applicationId}-<name>-repartition", where "applicationId" is
user-specified in StreamsConfig
via parameter
APPLICATION_ID_CONFIG
, "<name>" is an internally generated name, and
"-repartition" is a fixed suffix.
You can retrieve all generated internal topic names via Topology.describe()
.
Repartitioning can happen only for this KStream
but not for the provided KTable
.
For this case, all data of the stream will be redistributed through the repartitioning topic by writing all
records to it, and rereading all records from it, such that the join input KStream
is partitioned
correctly on its key.
leftJoin
in interface KStream<K,V>
VO
- the value type of the tableVR
- the value type of the result streamother
- the KTable
to be joined with this streamjoiner
- a ValueJoiner
that computes the join result for a pair of matching recordsKStream
that contains join-records for each key and values computed by the given
ValueJoiner
, one output for each input KStream
recordKStream.join(KTable, ValueJoiner, Joined)
,
KStream.leftJoin(GlobalKTable, KeyValueMapper, ValueJoiner)
public <KG,VG,VR> KStream<K,VR> join(GlobalKTable<KG,VG> globalTable, KeyValueMapper<? super K,? super V,? extends KG> keyMapper, ValueJoiner<? super V,? super VG,? extends VR> joiner)
KStream
GlobalKTable
's records using non-windowed inner equi join.
The join is a primary key table lookup join with join attribute
keyValueMapper.map(stream.keyValue) == table.key
.
"Table lookup join" means, that results are only computed if KStream
records are processed.
This is done by performing a lookup for matching records in the current internal GlobalKTable
state.
In contrast, processing GlobalKTable
input records will only update the internal GlobalKTable
state and will not produce any result records.
For each KStream
record that finds a corresponding record in GlobalKTable
the provided
ValueJoiner
will be called to compute a value (with arbitrary type) for the result record.
The key of the result record is the same as the key of this KStream
.
If a KStream
input record key or value is null
the record will not be included in the join
operation and thus no output record will be added to the resulting KStream
.
If keyValueMapper
returns null
implying no match exists, no output record will be added to the
resulting KStream
.
join
in interface KStream<K,V>
KG
- the key type of GlobalKTable
VG
- the value type of the GlobalKTable
VR
- the value type of the resulting KStream
globalTable
- the GlobalKTable
to be joined with this streamkeyMapper
- instance of KeyValueMapper
used to map from the (key, value) of this stream
to the key of the GlobalKTable
joiner
- a ValueJoiner
that computes the join result for a pair of matching recordsKStream
that contains join-records for each key and values computed by the given
ValueJoiner
, one output for each input KStream
recordKStream.leftJoin(GlobalKTable, KeyValueMapper, ValueJoiner)
public <KG,VG,VR> KStream<K,VR> leftJoin(GlobalKTable<KG,VG> globalTable, KeyValueMapper<? super K,? super V,? extends KG> keyMapper, ValueJoiner<? super V,? super VG,? extends VR> joiner)
KStream
GlobalKTable
's records using non-windowed left equi join.
In contrast to inner-join
, all records from this stream
will produce an output record (cf. below).
The join is a primary key table lookup join with join attribute
keyValueMapper.map(stream.keyValue) == table.key
.
"Table lookup join" means, that results are only computed if KStream
records are processed.
This is done by performing a lookup for matching records in the current internal GlobalKTable
state.
In contrast, processing GlobalKTable
input records will only update the internal GlobalKTable
state and will not produce any result records.
For each KStream
record whether or not it finds a corresponding record in GlobalKTable
the
provided ValueJoiner
will be called to compute a value (with arbitrary type) for the result record.
The key of the result record is the same as this KStream
.
If a KStream
input record key or value is null
the record will not be included in the join
operation and thus no output record will be added to the resulting KStream
.
If keyValueMapper
returns null
implying no match exists, a null
value will be
provided to ValueJoiner
.
If no GlobalKTable
record was found during lookup, a null
value will be provided to
ValueJoiner
.
leftJoin
in interface KStream<K,V>
KG
- the key type of GlobalKTable
VG
- the value type of the GlobalKTable
VR
- the value type of the resulting KStream
globalTable
- the GlobalKTable
to be joined with this streamkeyMapper
- instance of KeyValueMapper
used to map from the (key, value) of this stream
to the key of the GlobalKTable
joiner
- a ValueJoiner
that computes the join result for a pair of matching recordsKStream
that contains join-records for each key and values computed by the given
ValueJoiner
, one output for each input KStream
recordKStream.join(GlobalKTable, KeyValueMapper, ValueJoiner)
public <KR> KGroupedStream<KR,V> groupBy(KeyValueMapper<? super K,? super V,KR> selector)
KStream
KStream
on a new key that is selected using the provided KeyValueMapper
and default serializers and deserializers.
Grouping a stream on the record key is required before an aggregation operator can be applied to the data
(cf. KGroupedStream
).
The KeyValueMapper
selects a new key (which may or may not be of the same type) while preserving the original values.
If the new record key is null
the record will not be included in the resulting KGroupedStream
Because a new key is selected, an internal repartitioning topic may need to be created in Kafka if a
later operator depends on the newly selected key.
This topic will be named "${applicationId}-<name>-repartition", where "applicationId" is user-specified in
StreamsConfig
via parameter APPLICATION_ID_CONFIG
, "<name>" is
an internally generated name, and "-repartition" is a fixed suffix.
You can retrieve all generated internal topic names via Topology.describe()
.
All data of this stream will be redistributed through the repartitioning topic by writing all records to it,
and rereading all records from it, such that the resulting KGroupedStream
is partitioned on the new key.
This operation is equivalent to calling KStream.selectKey(KeyValueMapper)
followed by KStream.groupByKey()
.
If the key type is changed, it is recommended to use KStream.groupBy(KeyValueMapper, Serialized)
instead.
groupBy
in interface KStream<K,V>
KR
- the key type of the result KGroupedStream
selector
- a KeyValueMapper
that computes a new key for groupingKGroupedStream
that contains the grouped records of the original KStream
@Deprecated public <KR> KGroupedStream<KR,V> groupBy(KeyValueMapper<? super K,? super V,KR> selector, Serialized<KR,V> serialized)
KStream
KStream
on a new key that is selected using the provided KeyValueMapper
and Serde
s as specified by Serialized
.
Grouping a stream on the record key is required before an aggregation operator can be applied to the data
(cf. KGroupedStream
).
The KeyValueMapper
selects a new key (which may or may not be of the same type) while preserving the original values.
If the new record key is null
the record will not be included in the resulting KGroupedStream
.
Because a new key is selected, an internal repartitioning topic may need to be created in Kafka if a
later operator depends on the newly selected key.
This topic will be as "${applicationId}-<name>-repartition", where "applicationId" is user-specified in
StreamsConfig
via parameter APPLICATION_ID_CONFIG
, "<name>" is
an internally generated name, and "-repartition" is a fixed suffix.
You can retrieve all generated internal topic names via Topology.describe()
.
All data of this stream will be redistributed through the repartitioning topic by writing all records to it,
and rereading all records from it, such that the resulting KGroupedStream
is partitioned on the new key.
This operation is equivalent to calling KStream.selectKey(KeyValueMapper)
followed by KStream.groupByKey()
.
groupBy
in interface KStream<K,V>
KR
- the key type of the result KGroupedStream
selector
- a KeyValueMapper
that computes a new key for groupingKGroupedStream
that contains the grouped records of the original KStream
public <KR> KGroupedStream<KR,V> groupBy(KeyValueMapper<? super K,? super V,KR> selector, Grouped<KR,V> grouped)
KStream
KStream
on a new key that is selected using the provided KeyValueMapper
and Serde
s as specified by Grouped
.
Grouping a stream on the record key is required before an aggregation operator can be applied to the data
(cf. KGroupedStream
).
The KeyValueMapper
selects a new key (which may or may not be of the same type) while preserving the original values.
If the new record key is null
the record will not be included in the resulting KGroupedStream
.
Because a new key is selected, an internal repartitioning topic may need to be created in Kafka if a later
operator depends on the newly selected key.
This topic will be named "${applicationId}-<name>-repartition", where "applicationId" is user-specified in
StreamsConfig
via parameter APPLICATION_ID_CONFIG
, "<name>" is
either provided via Grouped.as(String)
or an internally generated name.
You can retrieve all generated internal topic names via Topology.describe()
.
All data of this stream will be redistributed through the repartitioning topic by writing all records to it,
and rereading all records from it, such that the resulting KGroupedStream
is partitioned on the new key.
This operation is equivalent to calling KStream.selectKey(KeyValueMapper)
followed by KStream.groupByKey()
.
groupBy
in interface KStream<K,V>
KR
- the key type of the result KGroupedStream
selector
- a KeyValueMapper
that computes a new key for groupinggrouped
- the Grouped
instance used to specify Serdes
and part of the name for a repartition topic if repartitioning is required.KGroupedStream
that contains the grouped records of the original KStream
public KGroupedStream<K,V> groupByKey()
KStream
KGroupedStream
while preserving the original values
and default serializers and deserializers.
Grouping a stream on the record key is required before an aggregation operator can be applied to the data
(cf. KGroupedStream
).
If a record key is null
the record will not be included in the resulting KGroupedStream
.
If a key changing operator was used before this operation (e.g., KStream.selectKey(KeyValueMapper)
,
KStream.map(KeyValueMapper)
, KStream.flatMap(KeyValueMapper)
, or
KStream.transform(TransformerSupplier, String...)
), and no data redistribution happened afterwards (e.g., via
KStream.through(String)
) an internal repartitioning topic may need to be created in Kafka if a later
operator depends on the newly selected key.
This topic will be named "${applicationId}-<name>-repartition", where "applicationId" is user-specified in
StreamsConfig
via parameter APPLICATION_ID_CONFIG
, "<name>" is
an internally generated name, and "-repartition" is a fixed suffix.
You can retrieve all generated internal topic names via Topology.describe()
.
For this case, all data of this stream will be redistributed through the repartitioning topic by writing all
records to it, and rereading all records from it, such that the resulting KGroupedStream
is partitioned
correctly on its key.
If the last key changing operator changed the key type, it is recommended to use
KStream.groupByKey(org.apache.kafka.streams.kstream.Grouped)
instead.
groupByKey
in interface KStream<K,V>
KGroupedStream
that contains the grouped records of the original KStream
KStream.groupBy(KeyValueMapper)
@Deprecated public KGroupedStream<K,V> groupByKey(Serialized<K,V> serialized)
KStream
KGroupedStream
while preserving the original values
and using the serializers as defined by Serialized
.
Grouping a stream on the record key is required before an aggregation operator can be applied to the data
(cf. KGroupedStream
).
If a record key is null
the record will not be included in the resulting KGroupedStream
.
If a key changing operator was used before this operation (e.g., KStream.selectKey(KeyValueMapper)
,
KStream.map(KeyValueMapper)
, KStream.flatMap(KeyValueMapper)
, or
KStream.transform(TransformerSupplier, String...)
), and no data redistribution happened afterwards (e.g., via
KStream.through(String)
) an internal repartitioning topic may need to be created in Kafka
if a later operator depends on the newly selected key.
This topic will be named "${applicationId}-<name>-repartition", where "applicationId" is user-specified in
StreamsConfig
via parameter APPLICATION_ID_CONFIG
, "<name>" is
an internally generated name, and "-repartition" is a fixed suffix.
You can retrieve all generated internal topic names via Topology.describe()
.
For this case, all data of this stream will be redistributed through the repartitioning topic by writing all
records to it, and rereading all records from it, such that the resulting KGroupedStream
is partitioned
correctly on its key.
groupByKey
in interface KStream<K,V>
KGroupedStream
that contains the grouped records of the original KStream
KStream.groupBy(KeyValueMapper)
public KGroupedStream<K,V> groupByKey(Grouped<K,V> grouped)
KStream
KGroupedStream
while preserving the original values
and using the serializers as defined by Grouped
.
Grouping a stream on the record key is required before an aggregation operator can be applied to the data
(cf. KGroupedStream
).
If a record key is null
the record will not be included in the resulting KGroupedStream
.
If a key changing operator was used before this operation (e.g., KStream.selectKey(KeyValueMapper)
,
KStream.map(KeyValueMapper)
, KStream.flatMap(KeyValueMapper)
, or
KStream.transform(TransformerSupplier, String...)
), and no data redistribution happened afterwards (e.g., via
KStream.through(String)
) an internal repartitioning topic may need to be created in Kafka if a later operator
depends on the newly selected key.
This topic will be named "${applicationId}-<name>-repartition", where "applicationId" is user-specified in
StreamsConfig
via parameter APPLICATION_ID_CONFIG
, <name> is
either provided via Grouped.as(String)
or an internally generated name,
and "-repartition" is a fixed suffix.
You can retrieve all generated internal topic names via Topology.describe()
.
For this case, all data of this stream will be redistributed through the repartitioning topic by writing all
records to it, and rereading all records from it, such that the resulting KGroupedStream
is partitioned
correctly on its key.
groupByKey
in interface KStream<K,V>
grouped
- the Grouped
instance used to specify Serdes
and part of the name for a repartition topic if repartitioning is required.KGroupedStream
that contains the grouped records of the original KStream
KStream.groupBy(KeyValueMapper)