Package 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 Detail

      • 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 interval.

        To query the local ReadOnlyKeyValueStore 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
 ReadOnlyKeyValueStore<K, ValueAndTimestamp<VOut>> localStore = streams.store(queryableStoreName, QueryableStoreTypes.<K, ValueAndTimestamp<VOut>> timestampedKeyValueStore());
 K key = "some-key";
 ValueAndTimestamp<VOut> 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 (which always will be of type TimestampedKeyValueStore) 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 interval.

        To query the local ReadOnlyKeyValueStore 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
 ReadOnlyKeyValueStore<K, ValueAndTimestamp<VOut>> localStore = streams.store(queryableStoreName, QueryableStoreTypes.<K, ValueAndTimestamp<VOut>> timestampedKeyValueStore());
 K key = "some-key";
 ValueAndTimestamp<VOut> 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 (which always will be of type TimestampedKeyValueStore) 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 interval.

        To query the local ReadOnlyKeyValueStore 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
 ReadOnlyKeyValueStore<K, ValueAndTimestamp<VOut>> localStore = streams.store(queryableStoreName, QueryableStoreTypes.<K, ValueAndTimestamp<VOut>> timestampedKeyValueStore());
 K key = "some-key";
 ValueAndTimestamp<VOut> 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 (which always will be of type TimestampedKeyValueStore -- regardless of what is specified in the parameter materialized) 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 interval.

        To query the local ReadOnlyKeyValueStore 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
 ReadOnlyKeyValueStore<K, ValueAndTimestamp<VOut>> localStore = streams.store(queryableStoreName, QueryableStoreTypes.<K, ValueAndTimestamp<VOut>> timestampedKeyValueStore());
 K key = "some-key";
 ValueAndTimestamp<VOut> 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 (which always will be of type TimestampedKeyValueStore -- regardless of what is specified in the parameter materialized) 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