Package org.apache.kafka.streams.kstream.internals

Class TimeWindowedKStreamImpl<K,​V,​W extends Window>

    • Method Detail

      • count

        public KTable<Windowed<K>,​Long> count()
        Description copied from interface: TimeWindowedKStream
        Count the number of records in this stream by the grouped key and defined windows. Records with null key or value are ignored.

        The result is written into a local WindowStore (which is basically an ever-updating materialized view). The default key serde from the config will be used for serializing the result. If a different serde is required then you should use TimeWindowedKStream.count(Materialized). Furthermore, updates to the store are sent downstream into a KTable changelog stream. Not all updates might get sent downstream, as an internal cache is used to deduplicate consecutive updates to the same window and 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.

        For failure and recovery the store will be backed by an internal changelog topic that will be created in Kafka. The changelog topic will be named "${applicationId}-${internalStoreName}-changelog", where "applicationId" is user-specified in StreamsConfig via parameter APPLICATION_ID_CONFIG, "internalStoreName" is an internal name and "-changelog" is a fixed suffix. Note that the internal store name may not be queryable through Interactive Queries.

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

        Specified by:
        count in interface TimeWindowedKStream<K,​V>
        Returns:
        a windowed KTable that contains "update" records with unmodified keys and Long values that represent the latest (rolling) count (i.e., number of records) for each key within a window

      • count

        public KTable<Windowed<K>,​Long> count​(Named named)
        Description copied from interface: TimeWindowedKStream
        Count the number of records in this stream by the grouped key and defined windows. Records with null key or value are ignored.

        The result is written into a local WindowStore (which is basically an ever-updating materialized view). The default key serde from the config will be used for serializing the result. If a different serde is required then you should use TimeWindowedKStream.count(Named, Materialized). Furthermore, updates to the store are sent downstream into a KTable changelog stream. Not all updates might get sent downstream, as an internal cache is used to deduplicate consecutive updates to the same window and 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.

        For failure and recovery the store will be backed by an internal changelog topic that will be created in Kafka. The changelog topic will be named "${applicationId}-${internalStoreName}-changelog", where "applicationId" is user-specified in StreamsConfig via parameter APPLICATION_ID_CONFIG, "internalStoreName" is an internal name and "-changelog" is a fixed suffix. Note that the internal store name may not be queryable through Interactive Queries.

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

        Specified by:
        count in interface TimeWindowedKStream<K,​V>
        Parameters:
        named - a Named config used to name the processor in the topology. Cannot be null.
        Returns:
        a windowed KTable that contains "update" records with unmodified keys and Long values that represent the latest (rolling) count (i.e., number of records) for each key within a window

      • count

        public KTable<Windowed<K>,​Long> count​(Materialized<K,​Long,​WindowStore<org.apache.kafka.common.utils.Bytes,​byte[]>> materialized)
        Description copied from interface: TimeWindowedKStream
        Count the number of records in this stream by the grouped key and defined windows. Records with null key or value are ignored.

        The result is written into a local WindowStore (which is basically an ever-updating materialized view) that can be queried using the name provided with Materialized. Furthermore, updates to the store are sent downstream into a KTable changelog stream.

        Not all updates might get sent downstream, as an internal cache will be used to deduplicate consecutive updates to the same window and key if caching is enabled on the Materialized instance. When caching is enabled 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 ReadOnlyWindowStore it must be obtained via KafkaStreams#store(...): 

        
 KafkaStreams streams = ... // counting words
 Store queryableStoreName = ... // the queryableStoreName should be the name of the store as defined by the Materialized instance
 ReadOnlyWindowStore<K, ValueAndTimestamp<Long>> localWindowStore = streams.store(queryableStoreName, QueryableStoreTypes.<K, ValueAndTimestamp<Long>>timestampedWindowStore());

 K key = "some-word";
 long fromTime = ...;
 long toTime = ...;
 WindowStoreIterator<ValueAndTimestamp<Long>> countForWordsForWindows = localWindowStore.fetch(key, timeFrom, timeTo); // 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().

        Specified by:
        count in interface TimeWindowedKStream<K,​V>
        Parameters:
        materialized - an instance of Materialized used to materialize a state store. Cannot be null. Note: the valueSerde will be automatically set to Serdes#Long() if there is no valueSerde provided
        Returns:
        a windowed KTable that contains "update" records with unmodified keys and Long values that represent the latest (rolling) count (i.e., number of records) for each key within a window

      • count

        public KTable<Windowed<K>,​Long> count​(Named named,
                                            Materialized<K,​Long,​WindowStore<org.apache.kafka.common.utils.Bytes,​byte[]>> materialized)
        Description copied from interface: TimeWindowedKStream
        Count the number of records in this stream by the grouped key and defined windows. Records with null key or value are ignored.

        The result is written into a local WindowStore (which is basically an ever-updating materialized view) that can be queried using the name provided with Materialized. Furthermore, updates to the store are sent downstream into a KTable changelog stream.

        Not all updates might get sent downstream, as an internal cache will be used to deduplicate consecutive updates to the same window and key if caching is enabled on the Materialized instance. When caching is enabled 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 ReadOnlyWindowStore it must be obtained via KafkaStreams#store(...): 

        
 KafkaStreams streams = ... // counting words
 Store queryableStoreName = ... // the queryableStoreName should be the name of the store as defined by the Materialized instance
 ReadOnlyWindowStore<K, ValueAndTimestamp<Long>> localWindowStore = streams.store(queryableStoreName, QueryableStoreTypes.<K, ValueAndTimestamp<Long>>timestampedWindowStore());

 K key = "some-word";
 long fromTime = ...;
 long toTime = ...;
 WindowStoreIterator<ValueAndTimestamp<Long>> countForWordsForWindows = localWindowStore.fetch(key, timeFrom, timeTo); // 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().

        Specified by:
        count in interface TimeWindowedKStream<K,​V>
        Parameters:
        named - a Named config used to name the processor in the topology. Cannot be null.
        materialized - an instance of Materialized used to materialize a state store. Cannot be null. Note: the valueSerde will be automatically set to Serdes#Long() if there is no valueSerde provided
        Returns:
        a windowed KTable that contains "update" records with unmodified keys and Long values that represent the latest (rolling) count (i.e., number of records) for each key within a window

      • aggregate

        public <VR> KTable<Windowed<K>,​VR> aggregate​(Initializer<VR> initializer,
                                                   Aggregator<? super K,​? super V,​VR> aggregator)
        Description copied from interface: TimeWindowedKStream
        Aggregate the values of records in this stream by the grouped key and defined windows. Records with null key or value are ignored. Aggregating is a generalization of combining via reduce(...) as it, for example, allows the result to have a different type than the input values. The result is written into a local WindowStore (which is basically an ever-updating materialized view). Furthermore, updates to the store are sent downstream into a KTable changelog stream.

        The specified Initializer is applied directly before the first input record (per key) in each window is processed to provide an initial intermediate aggregation result that is used to process the first record for the window (per key). The specified Aggregator is applied for each input record and computes a new aggregate using the current aggregate (or for the very first record using the intermediate aggregation result provided via the Initializer) and the record's value. Thus, aggregate() can be used to compute aggregate functions like count (c.f. TimeWindowedKStream.count()).

        The default key and value serde from the config will be used for serializing the result. If a different serde is required then you should use TimeWindowedKStream.aggregate(Initializer, Aggregator, Materialized). Not all updates might get sent downstream, as an internal cache is used to deduplicate consecutive updates to the same window and 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.

        For failure and recovery the store will be backed by an internal changelog topic that will be created in Kafka. The changelog topic will be named "${applicationId}-${internalStoreName}-changelog", where "applicationId" is user-specified in StreamsConfig via parameter APPLICATION_ID_CONFIG, "internalStoreName" is an internal name and "-changelog" is a fixed suffix. Note that the internal store name may not be queryable through Interactive Queries.

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

        Specified by:
        aggregate in interface TimeWindowedKStream<K,​V>
        Type Parameters:
        VR - the value type of the resulting KTable
        Parameters:
        initializer - an Initializer that computes an initial intermediate aggregation result. Cannot be null.
        aggregator - an Aggregator that computes a new aggregate result. Cannot be null.
        Returns:
        a windowed KTable that contains "update" records with unmodified keys, and values that represent the latest (rolling) aggregate for each key within a window

      • aggregate

        public <VR> KTable<Windowed<K>,​VR> aggregate​(Initializer<VR> initializer,
                                                   Aggregator<? super K,​? super V,​VR> aggregator,
                                                   Named named)
        Description copied from interface: TimeWindowedKStream
        Aggregate the values of records in this stream by the grouped key and defined windows. Records with null key or value are ignored. Aggregating is a generalization of combining via reduce(...) as it, for example, allows the result to have a different type than the input values. The result is written into a local WindowStore (which is basically an ever-updating materialized view). Furthermore, updates to the store are sent downstream into a KTable changelog stream.

        The specified Initializer is applied directly before the first input record (per key) in each window is processed to provide an initial intermediate aggregation result that is used to process the first record for the window (per key). The specified Aggregator is applied for each input record and computes a new aggregate using the current aggregate (or for the very first record using the intermediate aggregation result provided via the Initializer) and the record's value. Thus, aggregate() can be used to compute aggregate functions like count (c.f. TimeWindowedKStream.count()).

        The default key and value serde from the config will be used for serializing the result. If a different serde is required then you should use TimeWindowedKStream.aggregate(Initializer, Aggregator, Named, Materialized). Not all updates might get sent downstream, as an internal cache is used to deduplicate consecutive updates to the same window and 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.

        For failure and recovery the store will be backed by an internal changelog topic that will be created in Kafka. The changelog topic will be named "${applicationId}-${internalStoreName}-changelog", where "applicationId" is user-specified in StreamsConfig via parameter APPLICATION_ID_CONFIG, "internalStoreName" is an internal name and "-changelog" is a fixed suffix. Note that the internal store name may not be queryable through Interactive Queries.

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

        Specified by:
        aggregate in interface TimeWindowedKStream<K,​V>
        Type Parameters:
        VR - the value type of the resulting KTable
        Parameters:
        initializer - an Initializer that computes an initial intermediate aggregation result. Cannot be null.
        aggregator - an Aggregator that computes a new aggregate result. Cannot be null.
        named - a Named config used to name the processor in the topology. Cannot be null.
        Returns:
        a windowed KTable that contains "update" records with unmodified keys, and values that represent the latest (rolling) aggregate for each key within a window

      • aggregate

        public <VR> KTable<Windowed<K>,​VR> aggregate​(Initializer<VR> initializer,
                                                   Aggregator<? super K,​? super V,​VR> aggregator,
                                                   Materialized<K,​VR,​WindowStore<org.apache.kafka.common.utils.Bytes,​byte[]>> materialized)
        Description copied from interface: TimeWindowedKStream
        Aggregate the values of records in this stream by the grouped key and defined windows. Records with null key or value are ignored. Aggregating is a generalization of combining via reduce(...) as it, for example, allows the result to have a different type than the input values. The result is written into a local WindowStore (which is basically an ever-updating materialized view) that can be queried using the store name as provided with Materialized. Furthermore, updates to the store are sent downstream into a KTable changelog stream.

        The specified Initializer is applied directly before the first input record (per key) in each window is processed to provide an initial intermediate aggregation result that is used to process the first record for the window (per key). The specified Aggregator is applied for each input record and computes a new aggregate using the current aggregate (or for the very first record using the intermediate aggregation result provided via the Initializer) and the record's value. Thus, aggregate() can be used to compute aggregate functions like count (c.f. TimeWindowedKStream.count()).

        Not all updates might get sent downstream, as an internal cache is used to deduplicate consecutive updates to the same window and key if caching is enabled on the Materialized instance. When caching is enabled 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 ReadOnlyWindowStore it must be obtained via KafkaStreams#store(...): 

        
 KafkaStreams streams = ... // counting words
 Store queryableStoreName = ... // the queryableStoreName should be the name of the store as defined by the Materialized instance
 ReadOnlyWindowStore<K, ValueAndTimestamp<VR>> localWindowStore = streams.store(queryableStoreName, QueryableStoreTypes.<K, ValueAndTimestamp<VR>>timestampedWindowStore());

 K key = "some-word";
 long fromTime = ...;
 long toTime = ...;
 WindowStoreIterator<ValueAndTimestamp<VR>> aggregateStore = localWindowStore.fetch(key, timeFrom, timeTo); // 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().

        Specified by:
        aggregate in interface TimeWindowedKStream<K,​V>
        Type Parameters:
        VR - the value type of the resulting KTable
        Parameters:
        initializer - an Initializer that computes an initial intermediate aggregation result. Cannot be null.
        aggregator - an Aggregator that computes a new aggregate result. Cannot be null.
        materialized - a Materialized config used to materialize a state store. Cannot be null.
        Returns:
        a windowed KTable that contains "update" records with unmodified keys, and values that represent the latest (rolling) aggregate for each key within a window

      • aggregate

        public <VR> KTable<Windowed<K>,​VR> aggregate​(Initializer<VR> initializer,
                                                   Aggregator<? super K,​? super V,​VR> aggregator,
                                                   Named named,
                                                   Materialized<K,​VR,​WindowStore<org.apache.kafka.common.utils.Bytes,​byte[]>> materialized)
        Description copied from interface: TimeWindowedKStream
        Aggregate the values of records in this stream by the grouped key and defined windows. Records with null key or value are ignored. Aggregating is a generalization of combining via reduce(...) as it, for example, allows the result to have a different type than the input values. The result is written into a local WindowStore (which is basically an ever-updating materialized view) that can be queried using the store name as provided with Materialized. Furthermore, updates to the store are sent downstream into a KTable changelog stream.

        The specified Initializer is applied directly before the first input record (per key) in each window is processed to provide an initial intermediate aggregation result that is used to process the first record for the window (per key). The specified Aggregator is applied for each input record and computes a new aggregate using the current aggregate (or for the very first record using the intermediate aggregation result provided via the Initializer) and the record's value. Thus, aggregate() can be used to compute aggregate functions like count (c.f. TimeWindowedKStream.count()).

        Not all updates might get sent downstream, as an internal cache will be used to deduplicate consecutive updates to the same window and key if caching is enabled on the Materialized instance. When caching is enabled 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 ReadOnlyWindowStore it must be obtained via KafkaStreams#store(...): 

        
 KafkaStreams streams = ... // counting words
 Store queryableStoreName = ... // the queryableStoreName should be the name of the store as defined by the Materialized instance
 ReadOnlyWindowStore<K, ValueAndTimestamp<VR>> localWindowStore = streams.store(queryableStoreName, QueryableStoreTypes.<K, ValueAndTimestamp<VR>>timestampedWindowStore());

 K key = "some-word";
 long fromTime = ...;
 long toTime = ...;
 WindowStoreIterator<ValueAndTimestamp<VR>> aggregateStore = localWindowStore.fetch(key, timeFrom, timeTo); // 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().

        Specified by:
        aggregate in interface TimeWindowedKStream<K,​V>
        Type Parameters:
        VR - the value type of the resulting KTable
        Parameters:
        initializer - an Initializer that computes an initial intermediate aggregation result. Cannot be null.
        aggregator - an Aggregator that computes a new aggregate result. Cannot be null.
        named - a Named config used to name the processor in the topology. Cannot be null.
        materialized - a Materialized config used to materialize a state store. Cannot be null.
        Returns:
        a windowed KTable that contains "update" records with unmodified keys, and values that represent the latest (rolling) aggregate for each key within a window

      • reduce

        public KTable<Windowed<K>,​V> reduce​(Reducer<V> reducer)
        Description copied from interface: TimeWindowedKStream
        Combine the values of records in this stream by the grouped key and defined windows. Records with null key or value are ignored. Combining implies that the type of the aggregate result is the same as the type of the input value (c.f. TimeWindowedKStream.aggregate(Initializer, Aggregator)).

        The result is written into a local WindowStore (which is basically an ever-updating materialized view). Furthermore, updates to the store are sent downstream into a KTable changelog stream. The default key and value serde from the config will be used for serializing the result. If a different serde is required then you should use TimeWindowedKStream.reduce(Reducer, Materialized) .

        The value of the first record per window initialized the aggregation result. The specified Reducer is applied for each additional input record per window and computes a new aggregate using the current aggregate (first argument) and the record's value (second argument): 

        
 // At the example of a Reducer<Long>
 new Reducer<Long>() {
   public Long apply(Long aggValue, Long currValue) {
     return aggValue + currValue;
   }
 }
        Thus, reduce() can be used to compute aggregate functions like sum, min, or max.

        Not all updates might get sent downstream, as an internal cache is used to deduplicate consecutive updates to the same window and 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.

        For failure and recovery the store will be backed by an internal changelog topic that will be created in Kafka. The changelog topic will be named "${applicationId}-${internalStoreName}-changelog", where "applicationId" is user-specified in StreamsConfig via parameter APPLICATION_ID_CONFIG, "internalStoreName" is an internal name and "-changelog" is a fixed suffix.

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

        Specified by:
        reduce in interface TimeWindowedKStream<K,​V>
        Parameters:
        reducer - a Reducer that computes a new aggregate result. Cannot be null.
        Returns:
        a windowed KTable that contains "update" records with unmodified keys, and values that represent the latest (rolling) aggregate for each key within a window

      • reduce

        public KTable<Windowed<K>,​V> reduce​(Reducer<V> reducer,
                                          Named named)
        Description copied from interface: TimeWindowedKStream
        Combine the values of records in this stream by the grouped key and defined windows. Records with null key or value are ignored. Combining implies that the type of the aggregate result is the same as the type of the input value.

        The result is written into a local WindowStore (which is basically an ever-updating materialized view). Furthermore, updates to the store are sent downstream into a KTable changelog stream. The default key and value serde from the config will be used for serializing the result. If a different serde is required then you should use TimeWindowedKStream.reduce(Reducer, Named, Materialized) .

        The value of the first record per window initialized the aggregation result. The specified Reducer is applied for each additional input record per window and computes a new aggregate using the current aggregate (first argument) and the record's value (second argument): 

        
 // At the example of a Reducer<Long>
 new Reducer<Long>() {
   public Long apply(Long aggValue, Long currValue) {
     return aggValue + currValue;
   }
 }
        Thus, reduce() can be used to compute aggregate functions like sum, min, or max.

        Not all updates might get sent downstream, as an internal cache is used to deduplicate consecutive updates to the same window and 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.

        For failure and recovery the store will be backed by an internal changelog topic that will be created in Kafka. The changelog topic will be named "${applicationId}-${internalStoreName}-changelog", where "applicationId" is user-specified in StreamsConfig via parameter APPLICATION_ID_CONFIG, "internalStoreName" is an internal name and "-changelog" is a fixed suffix.

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

        Specified by:
        reduce in interface TimeWindowedKStream<K,​V>
        Parameters:
        reducer - a Reducer that computes a new aggregate result. Cannot be null.
        named - a Named config used to name the processor in the topology. Cannot be null.
        Returns:
        a windowed KTable that contains "update" records with unmodified keys, and values that represent the latest (rolling) aggregate for each key within a window

      • reduce

        public KTable<Windowed<K>,​V> reduce​(Reducer<V> reducer,
                                          Materialized<K,​V,​WindowStore<org.apache.kafka.common.utils.Bytes,​byte[]>> materialized)
        Description copied from interface: TimeWindowedKStream
        Combine the values of records in this stream by the grouped key and defined windows. Records with null key or value are ignored. Combining implies that the type of the aggregate result is the same as the type of the input value.

        The result is written into a local WindowStore (which is basically an ever-updating materialized view) that can be queried using the store name as provided with Materialized. Furthermore, updates to the store are sent downstream into a KTable changelog stream.

        The value of the first record per window initialized the aggregation result. The specified Reducer is applied for each additional input record per window and computes a new aggregate using the current aggregate (first argument) and the record's value (second argument): 

        
 // At the example of a Reducer<Long>
 new Reducer<Long>() {
   public Long apply(Long aggValue, Long currValue) {
     return aggValue + currValue;
   }
 }
        Thus, reduce() can be used to compute aggregate functions like sum, min, or max.

        Not all updates might get sent downstream, as an internal cache will be used to deduplicate consecutive updates to the same window and key if caching is enabled on the Materialized instance. When caching is enabled 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 ReadOnlyWindowStore it must be obtained via KafkaStreams#store(...): 

        
 KafkaStreams streams = ... // counting words
 Store queryableStoreName = ... // the queryableStoreName should be the name of the store as defined by the Materialized instance
 ReadOnlyWindowStore<K, ValueAndTimestamp<V>> localWindowStore = streams.store(queryableStoreName, QueryableStoreTypes.<K, ValueAndTimestamp<V>>timestampedWindowStore());

 K key = "some-word";
 long fromTime = ...;
 long toTime = ...;
 WindowStoreIterator<ValueAndTimestamp<V>> reduceStore = localWindowStore.fetch(key, timeFrom, timeTo); // 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().

        Specified by:
        reduce in interface TimeWindowedKStream<K,​V>
        Parameters:
        reducer - a Reducer that computes a new aggregate result. Cannot be null.
        materialized - a Materialized config used to materialize a state store. Cannot be null.
        Returns:
        a windowed KTable that contains "update" records with unmodified keys, and values that represent the latest (rolling) aggregate for each key within a window

      • reduce

        public KTable<Windowed<K>,​V> reduce​(Reducer<V> reducer,
                                          Named named,
                                          Materialized<K,​V,​WindowStore<org.apache.kafka.common.utils.Bytes,​byte[]>> materialized)
        Description copied from interface: TimeWindowedKStream
        Combine the values of records in this stream by the grouped key and defined windows. Records with null key or value are ignored. Combining implies that the type of the aggregate result is the same as the type of the input value.

        The result is written into a local WindowStore (which is basically an ever-updating materialized view) that can be queried using the store name as provided with Materialized. Furthermore, updates to the store are sent downstream into a KTable changelog stream.

        The value of the first record per window initialized the aggregation result. The specified Reducer is applied for each additional input record per window and computes a new aggregate using the current aggregate (first argument) and the record's value (second argument): 

        
 // At the example of a Reducer<Long>
 new Reducer<Long>() {
   public Long apply(Long aggValue, Long currValue) {
     return aggValue + currValue;
   }
 }
        Thus, reduce() can be used to compute aggregate functions like sum, min, or max.

        Not all updates might get sent downstream, as an internal cache will be used to deduplicate consecutive updates to the same window and key if caching is enabled on the Materialized instance. When caching is enabled 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 ReadOnlyWindowStore it must be obtained via KafkaStreams#store(...): 

        
 KafkaStreams streams = ... // counting words
 Store queryableStoreName = ... // the queryableStoreName should be the name of the store as defined by the Materialized instance
 ReadOnlyWindowStore<K, ValueAndTimestamp<V>> localWindowStore = streams.store(queryableStoreName, QueryableStoreTypes.<K, ValueAndTimestamp<V>>timestampedWindowStore());

 K key = "some-word";
 long fromTime = ...;
 long toTime = ...;
 WindowStoreIterator<ValueAndTimestamp<V>> reduceStore = localWindowStore.fetch(key, timeFrom, timeTo); // 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().

        Specified by:
        reduce in interface TimeWindowedKStream<K,​V>
        Parameters:
        reducer - a Reducer that computes a new aggregate result. Cannot be null.
        named - a Named config used to name the processor in the topology. Cannot be null.
        materialized - a Materialized config used to materialize a state store. Cannot be null.
        Returns:
        a windowed KTable that contains "update" records with unmodified keys, and values that represent the latest (rolling) aggregate for each key within a window