org.apache.kafka.streams.kstream

Interface KStream<K,V>

Type Parameters:

K - Type of keys

V - Type of values

@InterfaceStability.Unstable
public interface KStream<K,V>

KStream is an abstraction of a record stream of key-value pairs.

A KStream is either defined from one or multiple Kafka topics that are consumed message by message or the result of a KStream transformation. A KTable can also be converted into a KStream.

A KStream can be transformed record by record, joined with another KStream or KTable, or can be aggregated into a KTable.

See Also:

KTable

Method Summary

Methods

Modifier and Type	Method and Description
<T> KTable<K,T>	aggregateByKey(Initializer<T> initializer, Aggregator<K,V,T> aggregator, Serde<K> keySerde, Serde<T> aggValueSerde, String name) Aggregate values of this stream by key into a new instance of ever-updating KTable.
<T> KTable<K,T>	aggregateByKey(Initializer<T> initializer, Aggregator<K,V,T> aggregator, String name) Aggregate values of this stream by key into a new instance of ever-updating KTable with default serializers and deserializers.
<T,W extends Window> KTable<Windowed<K>,T>	aggregateByKey(Initializer<T> initializer, Aggregator<K,V,T> aggregator, Windows<W> windows) Aggregate values of this stream by key on a window basis into a new instance of windowed KTable with default serializers and deserializers.
<T,W extends Window> KTable<Windowed<K>,T>	aggregateByKey(Initializer<T> initializer, Aggregator<K,V,T> aggregator, Windows<W> windows, Serde<K> keySerde, Serde<T> aggValueSerde) Aggregate values of this stream by key on a window basis into a new instance of windowed KTable.
KStream<K,V>[]	branch(Predicate<K,V>... predicates) Creates an array of KStream from this stream by branching the elements in the original stream based on the supplied predicates.
KTable<K,Long>	countByKey(Serde<K> keySerde, String name) Count number of records of this stream by key into a new instance of ever-updating KTable.
KTable<K,Long>	countByKey(String name) Count number of records of this stream by key into a new instance of ever-updating KTable with default serializers and deserializers.
<W extends Window> KTable<Windowed<K>,Long>	countByKey(Windows<W> windows) Count number of records of this stream by key on a window basis into a new instance of windowed KTable with default serializers and deserializers.
<W extends Window> KTable<Windowed<K>,Long>	countByKey(Windows<W> windows, Serde<K> keySerde) Count number of records of this stream by key on a window basis into a new instance of windowed KTable.
KStream<K,V>	filter(Predicate<K,V> predicate) Create a new instance of KStream that consists of all elements of this stream which satisfy a predicate.
KStream<K,V>	filterNot(Predicate<K,V> predicate) Create a new instance of KStream that consists all elements of this stream which do not satisfy a predicate.
<K1,V1> KStream<K1,V1>	flatMap(KeyValueMapper<K,V,Iterable<KeyValue<K1,V1>>> mapper) Create a new instance of KStream by transforming each element in this stream into zero or more elements in the new stream.
<V1> KStream<K,V1>	flatMapValues(ValueMapper<V,Iterable<V1>> processor) Create a new instance of KStream by transforming the value of each element in this stream into zero or more values with the same key in the new stream.
void	foreach(ForeachAction<K,V> action) Perform an action on each element of KStream.
<V1,R> KStream<K,R>	join(KStream<K,V1> otherStream, ValueJoiner<V,V1,R> joiner, JoinWindows windows) Combine element values of this stream with another KStream's elements of the same key using windowed Inner Join with default serializers and deserializers.
<V1,R> KStream<K,R>	join(KStream<K,V1> otherStream, ValueJoiner<V,V1,R> joiner, JoinWindows windows, Serde<K> keySerde, Serde<V> thisValueSerde, Serde<V1> otherValueSerde) Combine element values of this stream with another KStream's elements of the same key using windowed Inner Join.
<V1,R> KStream<K,R>	leftJoin(KStream<K,V1> otherStream, ValueJoiner<V,V1,R> joiner, JoinWindows windows) Combine values of this stream with another KStream's elements of the same key using windowed Left Join with default serializers and deserializers.
<V1,R> KStream<K,R>	leftJoin(KStream<K,V1> otherStream, ValueJoiner<V,V1,R> joiner, JoinWindows windows, Serde<K> keySerde, Serde<V1> otherValueSerde) Combine values of this stream with another KStream's elements of the same key using windowed Left Join.
<V1,V2> KStream<K,V2>	leftJoin(KTable<K,V1> table, ValueJoiner<V,V1,V2> joiner) Combine values of this stream with KTable's elements of the same key using non-windowed Left Join.
<K1,V1> KStream<K1,V1>	map(KeyValueMapper<K,V,KeyValue<K1,V1>> mapper) Create a new instance of KStream by transforming each element in this stream into a different element in the new stream.
<V1> KStream<K,V1>	mapValues(ValueMapper<V,V1> mapper) Create a new instance of KStream by transforming the value of each element in this stream into a new value in the new stream.
<V1,R> KStream<K,R>	outerJoin(KStream<K,V1> otherStream, ValueJoiner<V,V1,R> joiner, JoinWindows windows) Combine values of this stream with another KStream's elements of the same key using windowed Outer Join with default serializers and deserializers.
<V1,R> KStream<K,R>	outerJoin(KStream<K,V1> otherStream, ValueJoiner<V,V1,R> joiner, JoinWindows windows, Serde<K> keySerde, Serde<V> thisValueSerde, Serde<V1> otherValueSerde) Combine values of this stream with another KStream's elements of the same key using windowed Outer Join.
void	print() Print the elements of this stream to System.out Implementors will need to override toString for keys and values that are not of type String, Integer etc to get meaningful information.
void	print(Serde<K> keySerde, Serde<V> valSerde) Print the elements of this stream to System.out
void	process(ProcessorSupplier<K,V> processorSupplier, String... stateStoreNames) Process all elements in this stream, one element at a time, by applying a Processor.
KTable<K,V>	reduceByKey(Reducer<V> reducer, Serde<K> keySerde, Serde<V> valueSerde, String name) Combine values of this stream by key into a new instance of ever-updating KTable.
KTable<K,V>	reduceByKey(Reducer<V> reducer, String name) Combine values of this stream by key into a new instance of ever-updating KTable with default serializers and deserializers.
<W extends Window> KTable<Windowed<K>,V>	reduceByKey(Reducer<V> reducer, Windows<W> windows) Combine values of this stream by key on a window basis into a new instance of windowed KTable with default serializers and deserializers.
<W extends Window> KTable<Windowed<K>,V>	reduceByKey(Reducer<V> reducer, Windows<W> windows, Serde<K> keySerde, Serde<V> valueSerde) Combine values of this stream by key on a window basis into a new instance of windowed KTable.
<K1> KStream<K1,V>	selectKey(KeyValueMapper<K,V,K1> mapper) Create a new key from the current key and value.
KStream<K,V>	through(Serde<K> keySerde, Serde<V> valSerde, StreamPartitioner<K,V> partitioner, String topic) Materialize this stream to a topic, also creates a new instance of KStream from the topic using a customizable StreamPartitioner to determine the distribution of records to partitions.
KStream<K,V>	through(Serde<K> keySerde, Serde<V> valSerde, String topic) Materialize this stream to a topic, also creates a new instance of KStream from the topic.
KStream<K,V>	through(StreamPartitioner<K,V> partitioner, String topic) Materialize this stream to a topic, also creates a new instance of KStream from the topic using default serializers and deserializers and a customizable StreamPartitioner to determine the distribution of records to partitions.
KStream<K,V>	through(String topic) Materialize this stream to a topic, also creates a new instance of KStream from the topic using default serializers and deserializers and producer's DefaultPartitioner.
void	to(Serde<K> keySerde, Serde<V> valSerde, StreamPartitioner<K,V> partitioner, String topic) Materialize this stream to a topic using a customizable StreamPartitioner to determine the distribution of records to partitions.
void	to(Serde<K> keySerde, Serde<V> valSerde, String topic) Materialize this stream to a topic.
void	to(StreamPartitioner<K,V> partitioner, String topic) Materialize this stream to a topic using default serializers specified in the config and a customizable StreamPartitioner to determine the distribution of records to partitions.
void	to(String topic) Materialize this stream to a topic using default serializers specified in the config and producer's DefaultPartitioner.
<K1,V1> KStream<K1,V1>	transform(TransformerSupplier<K,V,KeyValue<K1,V1>> transformerSupplier, String... stateStoreNames) Create a new KStream instance by applying a Transformer to all elements in this stream, one element at a time.
<R> KStream<K,R>	transformValues(ValueTransformerSupplier<V,R> valueTransformerSupplier, String... stateStoreNames) Create a new KStream instance by applying a ValueTransformer to all values in this stream, one element at a time.
void	writeAsText(String filePath) Write the elements of this stream to a file at the given path.
void	writeAsText(String filePath, Serde<K> keySerde, Serde<V> valSerde)

Method Detail

filter

KStream<K,V> filter(Predicate<K,V> predicate)

Create a new instance of KStream that consists of all elements of this stream which satisfy a predicate.

Parameters:

predicate - the instance of Predicate

Returns:

a KStream that contains only those records that satisfy the given predicate

filterNot

KStream<K,V> filterNot(Predicate<K,V> predicate)

Create a new instance of KStream that consists all elements of this stream which do not satisfy a predicate.

Parameters:

predicate - the instance of Predicate

Returns:

a KStream that contains only those records that do not satisfy the given predicate

selectKey

<K1> KStream<K1,V> selectKey(KeyValueMapper<K,V,K1> mapper)

Create a new key from the current key and value.

Type Parameters:

K1 - the new key type on the stream

Parameters:

mapper - the instance of KeyValueMapper

Returns:

a KStream that contains records with different key type and same value type

map

<K1,V1> KStream<K1,V1> map(KeyValueMapper<K,V,KeyValue<K1,V1>> mapper)

Create a new instance of KStream by transforming each element in this stream into a different element in the new stream.

Type Parameters:

K1 - the key type of the new stream

V1 - the value type of the new stream

Parameters:

mapper - the instance of KeyValueMapper

Returns:

a KStream that contains records with new key and value type

mapValues

<V1> KStream<K,V1> mapValues(ValueMapper<V,V1> mapper)

Create a new instance of KStream by transforming the value of each element in this stream into a new value in the new stream.

Type Parameters:

V1 - the value type of the new stream

Parameters:

mapper - the instance of ValueMapper

Returns:

a KStream that contains records with unmodified keys and new values of different type

print

void print()

Print the elements of this stream to System.out Implementors will need to override toString for keys and values that are not of type String, Integer etc to get meaningful information.

print

void print(Serde<K> keySerde,
         Serde<V> valSerde)

Print the elements of this stream to System.out

Parameters:

keySerde - key serde used to send key-value pairs, if not specified the default serde defined in the configs will be used

valSerde - value serde used to send key-value pairs, if not specified the default serde defined in the configs will be used Implementors will need to override toString for keys and values that are not of type String, Integer etc to get meaningful information.

writeAsText

void writeAsText(String filePath)

Write the elements of this stream to a file at the given path.

Parameters:

filePath - name of file to write to Implementors will need to override toString for keys and values that are not of type String, Integer etc to get meaningful information.

writeAsText

void writeAsText(String filePath,
               Serde<K> keySerde,
               Serde<V> valSerde)

Parameters:

filePath - name of file to write to

keySerde - key serde used to send key-value pairs, if not specified the default serde defined in the configs will be used

valSerde - value serde used to send key-value pairs, if not specified the default serde defined in the configs will be used Implementors will need to override toString for keys and values that are not of type String, Integer etc to get meaningful information.

flatMap

<K1,V1> KStream<K1,V1> flatMap(KeyValueMapper<K,V,Iterable<KeyValue<K1,V1>>> mapper)

Create a new instance of KStream by transforming each element in this stream into zero or more elements in the new stream.

Type Parameters:

K1 - the key type of the new stream

V1 - the value type of the new stream

Parameters:

mapper - the instance of KeyValueMapper

Returns:

a KStream that contains more or less records with new key and value type

flatMapValues

<V1> KStream<K,V1> flatMapValues(ValueMapper<V,Iterable<V1>> processor)

Create a new instance of KStream by transforming the value of each element in this stream into zero or more values with the same key in the new stream.

Type Parameters:

V1 - the value type of the new stream

Parameters:

processor - the instance of ValueMapper

Returns:

a KStream that contains more or less records with unmodified keys and new values of different type

branch

KStream<K,V>[] branch(Predicate<K,V>... predicates)

Creates an array of KStream from this stream by branching the elements in the original stream based on the supplied predicates. Each element 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: An element in the original stream is assigned to the corresponding result stream for the first predicate that evaluates to true, and assigned to this stream only. An element will be dropped if none of the predicates evaluate to true.

Parameters:

predicates - the ordered list of Predicate instances

Returns:

multiple distinct substreams of this KStream

through

KStream<K,V> through(String topic)

Materialize this stream to a topic, also creates a new instance of KStream from the topic using default serializers and deserializers and producer's DefaultPartitioner. This is equivalent to calling to(String) and KStreamBuilder.stream(String...).

Parameters:

topic - the topic name

Returns:

a KStream that contains the exact same records as this KStream

foreach

void foreach(ForeachAction<K,V> action)

Perform an action on each element of KStream. Note that this is a terminal operation that returns void.

Parameters:

action - an action to perform on each element

through

KStream<K,V> through(StreamPartitioner<K,V> partitioner,
                   String topic)

Materialize this stream to a topic, also creates a new instance of KStream from the topic using default serializers and deserializers and a customizable StreamPartitioner to determine the distribution of records to partitions. This is equivalent to calling to(StreamPartitioner, String) and KStreamBuilder.stream(String...).

Parameters:

partitioner - the function used to determine how records are distributed among partitions of the topic, if not specified producer's DefaultPartitioner will be used

topic - the topic name

Returns:

a KStream that contains the exact same records as this KStream

through

KStream<K,V> through(Serde<K> keySerde,
                   Serde<V> valSerde,
                   String topic)

Materialize this stream to a topic, also creates a new instance of KStream from the topic. If keySerde provides a org.apache.kafka.streams.kstream.internals.WindowedSerializer for the key org.apache.kafka.streams.kstream.internals.WindowedStreamPartitioner is used — otherwise producer's DefaultPartitioner is used. This is equivalent to calling to(Serde, Serde, String) and KStreamBuilder.stream(Serde, Serde, String...).

Parameters:

keySerde - key serde used to send key-value pairs, if not specified the default key serde defined in the configuration will be used

valSerde - value serde used to send key-value pairs, if not specified the default value serde defined in the configuration will be used

topic - the topic name

Returns:

a KStream that contains the exact same records as this KStream

through

KStream<K,V> through(Serde<K> keySerde,
                   Serde<V> valSerde,
                   StreamPartitioner<K,V> partitioner,
                   String topic)

Materialize this stream to a topic, also creates a new instance of KStream from the topic using a customizable StreamPartitioner to determine the distribution of records to partitions. This is equivalent to calling to(Serde, Serde, StreamPartitioner, String) and KStreamBuilder.stream(Serde, Serde, String...).

Parameters:

keySerde - key serde used to send key-value pairs, if not specified the default key serde defined in the configuration will be used

valSerde - value serde used to send key-value pairs, if not specified the default value serde defined in the configuration will be used

partitioner - the function used to determine how records are distributed among partitions of the topic, if not specified and keySerde provides a org.apache.kafka.streams.kstream.internals.WindowedSerializer for the key org.apache.kafka.streams.kstream.internals.WindowedStreamPartitioner will be used — otherwise DefaultPartitioner will be used

topic - the topic name

Returns:

a KStream that contains the exact same records as this KStream

to

void to(String topic)

Materialize this stream to a topic using default serializers specified in the config and producer's DefaultPartitioner.

Parameters:

topic - the topic name

to

void to(StreamPartitioner<K,V> partitioner,
      String topic)

Materialize this stream to a topic using default serializers specified in the config and a customizable StreamPartitioner to determine the distribution of records to partitions.

Parameters:

partitioner - the function used to determine how records are distributed among partitions of the topic, if not specified producer's DefaultPartitioner will be used

topic - the topic name

to

void to(Serde<K> keySerde,
      Serde<V> valSerde,
      String topic)

Materialize this stream to a topic. If keySerde provides a org.apache.kafka.streams.kstream.internals.WindowedSerializer for the key org.apache.kafka.streams.kstream.internals.WindowedStreamPartitioner is used — otherwise producer's DefaultPartitioner is used.

Parameters:

keySerde - key serde used to send key-value pairs, if not specified the default serde defined in the configs will be used

valSerde - value serde used to send key-value pairs, if not specified the default serde defined in the configs will be used

topic - the topic name

to

void to(Serde<K> keySerde,
      Serde<V> valSerde,
      StreamPartitioner<K,V> partitioner,
      String topic)

Materialize this stream to a topic using a customizable StreamPartitioner to determine the distribution of records to partitions.

Parameters:

keySerde - key serde used to send key-value pairs, if not specified the default serde defined in the configs will be used

valSerde - value serde used to send key-value pairs, if not specified the default serde defined in the configs will be used

partitioner - the function used to determine how records are distributed among partitions of the topic, if not specified and keySerde provides a org.apache.kafka.streams.kstream.internals.WindowedSerializer for the key org.apache.kafka.streams.kstream.internals.WindowedStreamPartitioner will be used — otherwise DefaultPartitioner will be used

topic - the topic name

transform

<K1,V1> KStream<K1,V1> transform(TransformerSupplier<K,V,KeyValue<K1,V1>> transformerSupplier,
                               String... stateStoreNames)

Create a new KStream instance by applying a Transformer to all elements in this stream, one element at a time.

Parameters:

transformerSupplier - the instance of TransformerSupplier that generates Transformer

stateStoreNames - the names of the state store used by the processor

Returns:

a new KStream with transformed key and value types

transformValues

<R> KStream<K,R> transformValues(ValueTransformerSupplier<V,R> valueTransformerSupplier,
                               String... stateStoreNames)

Create a new KStream instance by applying a ValueTransformer to all values in this stream, one element at a time.

Parameters:

valueTransformerSupplier - the instance of ValueTransformerSupplier that generates ValueTransformer

stateStoreNames - the names of the state store used by the processor

Returns:

a KStream that contains records with unmodified keys and transformed values with type R

process

void process(ProcessorSupplier<K,V> processorSupplier,
           String... stateStoreNames)

Process all elements in this stream, one element at a time, by applying a Processor.

Parameters:

processorSupplier - the supplier of ProcessorSupplier that generates Processor

stateStoreNames - the names of the state store used by the processor

join

<V1,R> KStream<K,R> join(KStream<K,V1> otherStream,
                       ValueJoiner<V,V1,R> joiner,
                       JoinWindows windows,
                       Serde<K> keySerde,
                       Serde<V> thisValueSerde,
                       Serde<V1> otherValueSerde)

Combine element values of this stream with another KStream's elements of the same key using windowed Inner Join.

Type Parameters:

V1 - the value type of the other stream

R - the value type of the new stream

Parameters:

otherStream - the instance of KStream joined with this stream

joiner - the instance of ValueJoiner

windows - the specification of the JoinWindows

keySerde - key serdes for materializing both streams, if not specified the default serdes defined in the configs will be used

thisValueSerde - value serdes for materializing this stream, if not specified the default serdes defined in the configs will be used

otherValueSerde - value serdes for materializing the other stream, if not specified the default serdes defined in the configs will be used

Returns:

a 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 intervals

join

<V1,R> KStream<K,R> join(KStream<K,V1> otherStream,
                       ValueJoiner<V,V1,R> joiner,
                       JoinWindows windows)

Combine element values of this stream with another KStream's elements of the same key using windowed Inner Join with default serializers and deserializers.

Type Parameters:

V1 - the value type of the other stream

R - the value type of the new stream

Parameters:

otherStream - the instance of KStream joined with this stream

joiner - the instance of ValueJoiner

windows - the specification of the JoinWindows

Returns:

a 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 intervals

outerJoin

<V1,R> KStream<K,R> outerJoin(KStream<K,V1> otherStream,
                            ValueJoiner<V,V1,R> joiner,
                            JoinWindows windows,
                            Serde<K> keySerde,
                            Serde<V> thisValueSerde,
                            Serde<V1> otherValueSerde)

Combine values of this stream with another KStream's elements of the same key using windowed Outer Join.

Type Parameters:

V1 - the value type of the other stream

R - the value type of the new stream

Parameters:

otherStream - the instance of KStream joined with this stream

joiner - the instance of ValueJoiner

windows - the specification of the JoinWindows

keySerde - key serdes for materializing both streams, if not specified the default serdes defined in the configs will be used

thisValueSerde - value serdes for materializing this stream, if not specified the default serdes defined in the configs will be used

otherValueSerde - value serdes for materializing the other stream, if not specified the default serdes defined in the configs will be used

Returns:

a 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 intervals

outerJoin

<V1,R> KStream<K,R> outerJoin(KStream<K,V1> otherStream,
                            ValueJoiner<V,V1,R> joiner,
                            JoinWindows windows)

Combine values of this stream with another KStream's elements of the same key using windowed Outer Join with default serializers and deserializers.

Type Parameters:

V1 - the value type of the other stream

R - the value type of the new stream

Parameters:

otherStream - the instance of KStream joined with this stream

joiner - the instance of ValueJoiner

windows - the specification of the JoinWindows

Returns:

a 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 intervals

leftJoin

<V1,R> KStream<K,R> leftJoin(KStream<K,V1> otherStream,
                           ValueJoiner<V,V1,R> joiner,
                           JoinWindows windows,
                           Serde<K> keySerde,
                           Serde<V1> otherValueSerde)

Combine values of this stream with another KStream's elements of the same key using windowed Left Join.

Type Parameters:

V1 - the value type of the other stream

R - the value type of the new stream

Parameters:

otherStream - the instance of KStream joined with this stream

joiner - the instance of ValueJoiner

windows - the specification of the JoinWindows

keySerde - key serdes for materializing the other stream, if not specified the default serdes defined in the configs will be used

otherValueSerde - value serdes for materializing the other stream, if not specified the default serdes defined in the configs will be used

Returns:

a 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 intervals

leftJoin

<V1,R> KStream<K,R> leftJoin(KStream<K,V1> otherStream,
                           ValueJoiner<V,V1,R> joiner,
                           JoinWindows windows)

Combine values of this stream with another KStream's elements of the same key using windowed Left Join with default serializers and deserializers.

Type Parameters:

V1 - the value type of the other stream

R - the value type of the new stream

Parameters:

otherStream - the instance of KStream joined with this stream

joiner - the instance of ValueJoiner

windows - the specification of the JoinWindows

Returns:

a 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 intervals

leftJoin

<V1,V2> KStream<K,V2> leftJoin(KTable<K,V1> table,
                             ValueJoiner<V,V1,V2> joiner)

Combine values of this stream with KTable's elements of the same key using non-windowed Left Join.

Type Parameters:

V1 - the value type of the table

V2 - the value type of the new stream

Parameters:

table - the instance of KTable joined with this stream

joiner - the instance of ValueJoiner

Returns:

a 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

reduceByKey

<W extends Window> KTable<Windowed<K>,V> reduceByKey(Reducer<V> reducer,
                                                   Windows<W> windows,
                                                   Serde<K> keySerde,
                                                   Serde<V> valueSerde)

Combine values of this stream by key on a window basis into a new instance of windowed KTable.

Parameters:

reducer - the instance of Reducer

windows - the specification of the aggregation Windows

keySerde - key serdes for materializing the aggregated table, if not specified the default serdes defined in the configs will be used

valueSerde - value serdes for materializing the aggregated table, if not specified the default serdes defined in the configs will be used

Returns:

a windowed KTable which can be treated as a list of KTables where each table contains records with unmodified keys and values that represent the latest (rolling) aggregate for each key within that window

reduceByKey

<W extends Window> KTable<Windowed<K>,V> reduceByKey(Reducer<V> reducer,
                                                   Windows<W> windows)

Combine values of this stream by key on a window basis into a new instance of windowed KTable with default serializers and deserializers.

Parameters:

reducer - the instance of Reducer

windows - the specification of the aggregation Windows

Returns:

a windowed KTable which can be treated as a list of KTables where each table contains records with unmodified keys and values that represent the latest (rolling) aggregate for each key within that window

reduceByKey

KTable<K,V> reduceByKey(Reducer<V> reducer,
                      Serde<K> keySerde,
                      Serde<V> valueSerde,
                      String name)

Combine values of this stream by key into a new instance of ever-updating KTable.

Parameters:

reducer - the instance of Reducer

keySerde - key serdes for materializing the aggregated table, if not specified the default serdes defined in the configs will be used

valueSerde - value serdes for materializing the aggregated table, if not specified the default serdes defined in the configs will be used

name - the name of the resulted KTable

Returns:

a KTable that contains records with unmodified keys and values that represent the latest (rolling) aggregate for each key

reduceByKey

KTable<K,V> reduceByKey(Reducer<V> reducer,
                      String name)

Combine values of this stream by key into a new instance of ever-updating KTable with default serializers and deserializers.

Parameters:

reducer - the instance of Reducer

name - the name of the resulted KTable

Returns:

a KTable that contains records with unmodified keys and values that represent the latest (rolling) aggregate for each key

aggregateByKey

<T,W extends Window> KTable<Windowed<K>,T> aggregateByKey(Initializer<T> initializer,
                                                        Aggregator<K,V,T> aggregator,
                                                        Windows<W> windows,
                                                        Serde<K> keySerde,
                                                        Serde<T> aggValueSerde)

Aggregate values of this stream by key on a window basis into a new instance of windowed KTable.

Type Parameters:

T - the value type of the resulted KTable

Parameters:

initializer - the instance of Initializer

aggregator - the instance of Aggregator

windows - the specification of the aggregation Windows

keySerde - key serdes for materializing the aggregated table, if not specified the default serdes defined in the configs will be used

aggValueSerde - aggregate value serdes for materializing the aggregated table, if not specified the default serdes defined in the configs will be used

Returns:

a windowed KTable which can be treated as a list of KTables where each table contains records with unmodified keys and values with type T that represent the latest (rolling) aggregate for each key within that window

aggregateByKey

<T,W extends Window> KTable<Windowed<K>,T> aggregateByKey(Initializer<T> initializer,
                                                        Aggregator<K,V,T> aggregator,
                                                        Windows<W> windows)

Aggregate values of this stream by key on a window basis into a new instance of windowed KTable with default serializers and deserializers.

Type Parameters:

T - the value type of the resulted KTable

Parameters:

initializer - the instance of Initializer

aggregator - the instance of Aggregator

windows - the specification of the aggregation Windows

Returns:

a windowed KTable which can be treated as a list of KTables where each table contains records with unmodified keys and values with type T that represent the latest (rolling) aggregate for each key within that window

aggregateByKey

<T> KTable<K,T> aggregateByKey(Initializer<T> initializer,
                             Aggregator<K,V,T> aggregator,
                             Serde<K> keySerde,
                             Serde<T> aggValueSerde,
                             String name)

Aggregate values of this stream by key into a new instance of ever-updating KTable.

Type Parameters:

T - the value type of the resulted KTable

Parameters:

initializer - the class of Initializer

aggregator - the class of Aggregator

keySerde - key serdes for materializing the aggregated table, if not specified the default serdes defined in the configs will be used

aggValueSerde - aggregate value serdes for materializing the aggregated table, if not specified the default serdes defined in the configs will be used

name - the name of the resulted KTable

Returns:

a KTable that contains records with unmodified keys and values (of different type) that represent the latest (rolling) aggregate for each key

aggregateByKey

<T> KTable<K,T> aggregateByKey(Initializer<T> initializer,
                             Aggregator<K,V,T> aggregator,
                             String name)

Aggregate values of this stream by key into a new instance of ever-updating KTable with default serializers and deserializers.

Type Parameters:

T - the value type of the resulted KTable

Parameters:

initializer - the class of Initializer

aggregator - the class of Aggregator

name - the name of the resulted KTable

Returns:

a KTable that contains records with unmodified keys and values (of different type) that represent the latest (rolling) aggregate for each key

countByKey

<W extends Window> KTable<Windowed<K>,Long> countByKey(Windows<W> windows,
                                                     Serde<K> keySerde)

Count number of records of this stream by key on a window basis into a new instance of windowed KTable.

Parameters:

windows - the specification of the aggregation Windows

keySerde - key serdes for materializing the counting table, if not specified the default serdes defined in the configs will be used

Returns:

a windowed KTable which can be treated as a list of KTables where each table contains records with unmodified keys and values that represent the latest (rolling) count (i.e., number of records) for each key within that window

countByKey

<W extends Window> KTable<Windowed<K>,Long> countByKey(Windows<W> windows)

Count number of records of this stream by key on a window basis into a new instance of windowed KTable with default serializers and deserializers.

Parameters:

windows - the specification of the aggregation Windows

Returns:

a windowed KTable which can be treated as a list of KTables where each table contains records with unmodified keys and values that represent the latest (rolling) count (i.e., number of records) for each key within that window

countByKey

KTable<K,Long> countByKey(Serde<K> keySerde,
                        String name)

Count number of records of this stream by key into a new instance of ever-updating KTable.

Parameters:

keySerde - key serdes for materializing the counting table, if not specified the default serdes defined in the configs will be used

name - the name of the resulted KTable

Returns:

a KTable that contains records with unmodified keys and values that represent the latest (rolling) count (i.e., number of records) for each key

countByKey

KTable<K,Long> countByKey(String name)

Count number of records of this stream by key into a new instance of ever-updating KTable with default serializers and deserializers.

Parameters:

name - the name of the resulted KTable

Returns:

a KTable that contains records with unmodified keys and values that represent the latest (rolling) count (i.e., number of records) for each key