org.apache.kafka.streams.kstream

Interface CogroupedKStream<K,VOut>

Type Parameters:

K - Type of keys

VOut - Type of values after agg

public interface CogroupedKStream<K,VOut>

CogroupedKStream is an abstraction of multiple grouped record streams of KeyValue pairs.

It is an intermediate representation after a grouping of KStreams, before the aggregations are applied to the new partitions resulting in a KTable.

A CogroupedKStream must be obtained from a KGroupedStream via cogroup(...).

Method Summary

Modifier and Type	Method and Description
KTable<K,VOut>	aggregate(Initializer<VOut> initializer) Aggregate the values of records in these streams by the grouped key.
KTable<K,VOut>	aggregate(Initializer<VOut> initializer, Materialized<K,VOut,KeyValueStore<org.apache.kafka.common.utils.Bytes,byte[]>> materialized) Aggregate the values of records in these streams by the grouped key.
KTable<K,VOut>	aggregate(Initializer<VOut> initializer, Named named) Aggregate the values of records in these streams by the grouped key.
KTable<K,VOut>	aggregate(Initializer<VOut> initializer, Named named, Materialized<K,VOut,KeyValueStore<org.apache.kafka.common.utils.Bytes,byte[]>> materialized) Aggregate the values of records in these streams by the grouped key.
<VIn> CogroupedKStream<K,VOut>	cogroup(KGroupedStream<K,VIn> groupedStream, Aggregator<? super K,? super VIn,VOut> aggregator) Add an already grouped KStream to this CogroupedKStream.
SessionWindowedCogroupedKStream<K,VOut>	windowedBy(SessionWindows windows) Create a new SessionWindowedCogroupedKStream instance that can be used to perform session windowed aggregations.
<W extends Window> TimeWindowedCogroupedKStream<K,VOut>	windowedBy(Windows<W> windows) Create a new TimeWindowedCogroupedKStream instance that can be used to perform windowed aggregations.

Method Detail

cogroup

<VIn> CogroupedKStream<K,VOut> cogroup(KGroupedStream<K,VIn> groupedStream,
                                       Aggregator<? super K,? super VIn,VOut> aggregator)

Add an already grouped KStream to this CogroupedKStream.

The added grouped KStream must have the same number of partitions as all existing streams of this CogroupedKStream. If this is not the case, you would need to call KStream.through(String) before grouping the KStream, using a pre-created topic with the "correct" number of partitions.

The specified Aggregator is applied in the actual aggregation step for each input record and computes a new aggregate using the current aggregate (or for the very first record per key using the initial intermediate aggregation result provided via the Initializer that is passed into aggregate(Initializer)) and the record's value.

Type Parameters:

VIn - Type of input values

Parameters:

groupedStream - a group stream

aggregator - an Aggregator that computes a new aggregate result

Returns:

a CogroupedKStream

aggregate

KTable<K,VOut> aggregate(Initializer<VOut> initializer)

Aggregate the values of records in these streams by the grouped key. Records with null key or value are ignored. The result is written into a local KeyValueStore (which is basically an ever-updating materialized view) that can be queried by the given store name in materialized. Furthermore, updates to the store are sent downstream into a KTable changelog stream.

To compute the aggregation the corresponding Aggregator as specified in cogroup(...) is used per input stream. The specified Initializer is applied once per key, directly before the first input record per key is processed to provide an initial intermediate aggregation result that is used to process the first record.

Not all updates might get sent downstream, as an internal cache is used to deduplicate consecutive updates to the same key. The rate of propagated updates depends on your input data rate, the number of distinct keys, the number of parallel running Kafka Streams instances, and the configuration parameters for cache size, and commit intervall.

To query the local KeyValueStore it must be obtained via KafkaStreams#store(...):

 KafkaStreams streams = ... // some aggregation on value type double
 String queryableStoreName = "storeName" // the store name should be the name of the store as defined by the Materialized instance
 KeyValueStore<String, Long> localStore = streams.store(queryableStoreName, QueryableStoreTypes.<String, Long>keyValueStore());
 String key = "some-key";
 Long aggForKey = localStore.get(key); // key must be local (application state is shared over all running Kafka Streams instances)
 

For non-local keys, a custom RPC mechanism must be implemented using KafkaStreams.allMetadata() to query the value of the key on a parallel running instance of your Kafka Streams application.

For failure and recovery the store will be backed by an internal changelog topic that will be created in Kafka. Therefore, the store name defined by the Materialized instance must be a valid Kafka topic name and cannot contain characters other than ASCII alphanumerics, '.', '_' and '-'. The changelog topic will be named "${applicationId}-${storeName}-changelog", where "applicationId" is user-specified in StreamsConfig via parameter APPLICATION_ID_CONFIG, "storeName" is a generated value, and "-changelog" is a fixed suffix.

You can retrieve all generated internal topic names via Topology.describe().

Parameters:

initializer - an Initializer that computes an initial intermediate aggregation result. Cannot be null.

Returns:

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

aggregate

KTable<K,VOut> aggregate(Initializer<VOut> initializer,
                         Named named)

Aggregate the values of records in these streams by the grouped key. Records with null key or value are ignored. The result is written into a local KeyValueStore (which is basically an ever-updating materialized view) that can be queried by the given store name in materialized. Furthermore, updates to the store are sent downstream into a KTable changelog stream.

To compute the aggregation the corresponding Aggregator as specified in cogroup(...) is used per input stream. The specified Initializer is applied once per key, directly before the first input record per key is processed to provide an initial intermediate aggregation result that is used to process the first record. The specified Named is applied once to the processor combining the grouped streams.

Not all updates might get sent downstream, as an internal cache is used to deduplicate consecutive updates to the same key. The rate of propagated updates depends on your input data rate, the number of distinct keys, the number of parallel running Kafka Streams instances, and the configuration parameters for cache size, and commit intervall.

To query the local KeyValueStore it must be obtained via KafkaStreams#store(...):

 KafkaStreams streams = ... // some aggregation on value type double
 String queryableStoreName = "storeName" // the store name should be the name of the store as defined by the Materialized instance
 KeyValueStore<String, Long> localStore = streams.store(queryableStoreName, QueryableStoreTypes.<String, Long>keyValueStore());
 String key = "some-key";
 Long aggForKey = localStore.get(key); // key must be local (application state is shared over all running Kafka Streams instances)
 

For non-local keys, a custom RPC mechanism must be implemented using KafkaStreams.allMetadata() to query the value of the key on a parallel running instance of your Kafka Streams application.

For failure and recovery the store will be backed by an internal changelog topic that will be created in Kafka. Therefore, the store name defined by the Materialized instance must be a valid Kafka topic name and cannot contain characters other than ASCII alphanumerics, '.', '_' and '-'. The changelog topic will be named "${applicationId}-${storeName}-changelog", where "applicationId" is user-specified in StreamsConfig via parameter APPLICATION_ID_CONFIG, "storeName" is the provide store name defined in Materialized, and "-changelog" is a fixed suffix.

You can retrieve all generated internal topic names via Topology.describe().

Parameters:

initializer - an Initializer that computes an initial intermediate aggregation result. Cannot be null.

named - name the processor. Cannot be null.

Returns:

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

aggregate

KTable<K,VOut> aggregate(Initializer<VOut> initializer,
                         Materialized<K,VOut,KeyValueStore<org.apache.kafka.common.utils.Bytes,byte[]>> materialized)

Aggregate the values of records in these streams by the grouped key. Records with null key or value are ignored. The result is written into a local KeyValueStore (which is basically an ever-updating materialized view) that can be queried by the given store name in materialized. Furthermore, updates to the store are sent downstream into a KTable changelog stream.

To compute the aggregation the corresponding Aggregator as specified in cogroup(...) is used per input stream. The specified Initializer is applied once per key, directly before the first input record per key is processed to provide an initial intermediate aggregation result that is used to process the first record.

Not all updates might get sent downstream, as an internal cache is used to deduplicate consecutive updates to the same key. The rate of propagated updates depends on your input data rate, the number of distinct keys, the number of parallel running Kafka Streams instances, and the configuration parameters for cache size, and commit intervall.

To query the local KeyValueStore it must be obtained via KafkaStreams#store(...):

 KafkaStreams streams = ... // some aggregation on value type double
 String queryableStoreName = "storeName" // the store name should be the name of the store as defined by the Materialized instance
 KeyValueStore localStore = streams.store(queryableStoreName, QueryableStoreTypes.keyValueStore());
 String key = "some-key";
 Long aggForKey = localStore.get(key); // key must be local (application state is shared over all running Kafka Streams instances)
 }

For non-local keys, a custom RPC mechanism must be implemented using KafkaStreams.allMetadata() to query the value of the key on a parallel running instance of your Kafka Streams application.

For failure and recovery the store will be backed by an internal changelog topic that will be created in Kafka. Therefore, the store name defined by the Materialized instance must be a valid Kafka topic name and cannot contain characters other than ASCII alphanumerics, '.', '_' and '-'. The changelog topic will be named "${applicationId}-${storeName}-changelog", where "applicationId" is user-specified in StreamsConfig via parameter APPLICATION_ID_CONFIG, "storeName" is the provide store name defined in Materialized, and "-changelog" is a fixed suffix.

You can retrieve all generated internal topic names via Topology.describe().

Parameters:

initializer - an Initializer that computes an initial intermediate aggregation result. Cannot be null.

materialized - an instance of Materialized used to materialize a state store. Cannot be null.

Returns:

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

aggregate

KTable<K,VOut> aggregate(Initializer<VOut> initializer,
                         Named named,
                         Materialized<K,VOut,KeyValueStore<org.apache.kafka.common.utils.Bytes,byte[]>> materialized)

Aggregate the values of records in these streams by the grouped key. Records with null key or value are ignored. The result is written into a local KeyValueStore (which is basically an ever-updating materialized view) that can be queried by the given store name in materialized. Furthermore, updates to the store are sent downstream into a KTable changelog stream.

To compute the aggregation the corresponding Aggregator as specified in cogroup(...) is used per input stream. The specified Initializer is applied once per key, directly before the first input record per key is processed to provide an initial intermediate aggregation result that is used to process the first record. The specified Named is used to name the processor combining the grouped streams.

Not all updates might get sent downstream, as an internal cache is used to deduplicate consecutive updates to the same key. The rate of propagated updates depends on your input data rate, the number of distinct keys, the number of parallel running Kafka Streams instances, and the configuration parameters for cache size, and commit intervall.

To query the local KeyValueStore it must be obtained via KafkaStreams.store(StoreQueryParameters) KafkaStreams#store(...)}:

 KafkaStreams streams = ... // some aggregation on value type double
 String queryableStoreName = "storeName" // the store name should be the name of the store as defined by the Materialized instance
 KeyValueStore localStore = streams.store(queryableStoreName, QueryableStoreTypes.keyValueStore());
 String key = "some-key";
 Long aggForKey = localStore.get(key); // key must be local (application state is shared over all running Kafka Streams instances)
 }

For non-local keys, a custom RPC mechanism must be implemented using KafkaStreams.allMetadata() to query the value of the key on a parallel running instance of your Kafka Streams application.

For failure and recovery the store will be backed by an internal changelog topic that will be created in Kafka. Therefore, the store name defined by the Materialized instance must be a valid Kafka topic name and cannot contain characters other than ASCII alphanumerics, '.', '_' and '-'. The changelog topic will be named "${applicationId}-${storeName}-changelog", where "applicationId" is user-specified in StreamsConfig via parameter APPLICATION_ID_CONFIG, "storeName" is the provide store name defined in Materialized, and "-changelog" is a fixed suffix.

You can retrieve all generated internal topic names via Topology.describe().

Parameters:

initializer - an Initializer that computes an initial intermediate aggregation result. Cannot be null.

materialized - an instance of Materialized used to materialize a state store. Cannot be null.

named - name the processors. Cannot be null.

Returns:

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

windowedBy

<W extends Window> TimeWindowedCogroupedKStream<K,VOut> windowedBy(Windows<W> windows)

Create a new TimeWindowedCogroupedKStream instance that can be used to perform windowed aggregations.

Type Parameters:

W - the window type

Parameters:

windows - the specification of the aggregation Windows

Returns:

an instance of TimeWindowedCogroupedKStream

windowedBy

SessionWindowedCogroupedKStream<K,VOut> windowedBy(SessionWindows windows)

Create a new SessionWindowedCogroupedKStream instance that can be used to perform session windowed aggregations.

Parameters:

windows - the specification of the aggregation SessionWindows

Returns:

an instance of SessionWindowedCogroupedKStream