Interface SessionWindowedKStream<K,V>
-
- Type Parameters:
K
- Type of keysV
- Type of values
- All Known Implementing Classes:
SessionWindowedKStreamImpl
public interface SessionWindowedKStream<K,V>
SessionWindowedKStream
is an abstraction of a windowed record stream ofKeyValue
pairs. It is an intermediate representation after a grouping and windowing of aKStream
before an aggregation is applied to the new (partitioned) windows resulting in a windowedKTable
(awindowed KTable
is aKTable
with key typeWindowed
.SessionWindows
are dynamic data driven windows. They have no fixed time boundaries, rather the size of the window is determined by the records. Please seeSessionWindows
for more details.New events are added to
SessionWindows
until their grace period ends (seeSessionWindows.grace(Duration)
). Furthermore, updates are sent downstream into a windowedKTable
changelog stream, where "windowed" implies that theKTable
key is a combined key of the original record key and a window ID.A
SessionWindowedKStream
must be obtained from aKGroupedStream
viaKGroupedStream.windowedBy(SessionWindows)
.- See Also:
KStream
,KGroupedStream
,SessionWindows
-
-
Method Summary
All Methods Instance Methods Abstract Methods Modifier and Type Method Description <VR> KTable<Windowed<K>,VR>
aggregate(Initializer<VR> initializer, Aggregator<? super K,? super V,VR> aggregator, Merger<? super K,VR> sessionMerger)
Aggregate the values of records in this stream by the grouped key and definedSessionWindows
.<VR> KTable<Windowed<K>,VR>
aggregate(Initializer<VR> initializer, Aggregator<? super K,? super V,VR> aggregator, Merger<? super K,VR> sessionMerger, Materialized<K,VR,SessionStore<org.apache.kafka.common.utils.Bytes,byte[]>> materialized)
Aggregate the values of records in this stream by the grouped key and definedSessionWindows
.KTable<Windowed<K>,java.lang.Long>
count()
Count the number of records in this stream by the grouped key intoSessionWindows
.KTable<Windowed<K>,java.lang.Long>
count(Materialized<K,java.lang.Long,SessionStore<org.apache.kafka.common.utils.Bytes,byte[]>> materialized)
Count the number of records in this stream by the grouped key intoSessionWindows
.KTable<Windowed<K>,V>
reduce(Reducer<V> reducer)
Combine values of this stream by the grouped key intoSessionWindows
.KTable<Windowed<K>,V>
reduce(Reducer<V> reducer, Materialized<K,V,SessionStore<org.apache.kafka.common.utils.Bytes,byte[]>> materializedAs)
Combine values of this stream by the grouped key intoSessionWindows
.
-
-
-
Method Detail
-
count
KTable<Windowed<K>,java.lang.Long> count()
Count the number of records in this stream by the grouped key intoSessionWindows
. Records withnull
key or value are ignored.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 forcache size
, andcommit intervall
.- Returns:
- a windowed
KTable
that contains "update" records with unmodified keys andLong
values that represent the latest (rolling) count (i.e., number of records) for each key within a window
-
count
KTable<Windowed<K>,java.lang.Long> count(Materialized<K,java.lang.Long,SessionStore<org.apache.kafka.common.utils.Bytes,byte[]>> materialized)
Count the number of records in this stream by the grouped key intoSessionWindows
. Records withnull
key or value are ignored. The result is written into a localSessionStore
(which is basically an ever-updating materialized view) that can be queried using the name provided withMaterialized
.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 theconfiguration
parameters forcache size
, andcommit intervall
To query the local windowed
KeyValueStore
it must be obtained viaKafkaStreams#store(...)
.
For non-local keys, a custom RPC mechanism must be implemented usingKafkaStreams streams = ... // compute sum Sting queryableStoreName = ... // the queryableStoreName should be the name of the store as defined by the Materialized instance ReadOnlySessionStore<String,Long> localWindowStore = streams.store(queryableStoreName, QueryableStoreTypes.<String, Long>ReadOnlySessionStore<String, Long>); String key = "some-key"; KeyValueIterator<Windowed<String>, Long> sumForKeyForWindows = localWindowStore.fetch(key); // key must be local (application state is shared over all running Kafka Streams instances)
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 parameterAPPLICATION_ID_CONFIG
, "storeName" is the provide store name defined inMaterialized
, and "-changelog" is a fixed suffix. You can retrieve all generated internal topic names viaTopology.describe()
.- Parameters:
materialized
- an instance ofMaterialized
used to materialize a state store. Cannot benull
. Note: the valueSerde will be automatically set toSerdes.Long()
if there is no valueSerde provided- Returns:
- a windowed
KTable
that contains "update" records with unmodified keys andLong
values that represent the latest (rolling) count (i.e., number of records) for each key within a window
-
aggregate
<VR> KTable<Windowed<K>,VR> aggregate(Initializer<VR> initializer, Aggregator<? super K,? super V,VR> aggregator, Merger<? super K,VR> sessionMerger)
Aggregate the values of records in this stream by the grouped key and definedSessionWindows
. Records withnull
key or value are ignored. Aggregating is a generalization ofcombining via reduce(...)
as it, for example, allows the result to have a different type than the input values.The specified
Initializer
is applied once per session directly before the first input record is processed to provide an initial intermediate aggregation result that is used to process the first record. The specifiedAggregator
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 theInitializer
) and the record's value. The specifiedMerger
is used to merge 2 existing sessions into one, i.e., when the windows overlap, they are merged into a single session and the old sessions are discarded. Thus,aggregate(Initializer, Aggregator, Merger)
can be used to compute aggregate functions like count (c.f.count()
)The default value serde from config will be used for serializing the result. If a different serde is required then you should use
aggregate(Initializer, Aggregator, Merger, 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 forcache size
, andcommit intervall
.- Type Parameters:
VR
- the value type of the resultingKTable
- Parameters:
initializer
- the instance ofInitializer
aggregator
- the instance ofAggregator
sessionMerger
- the instance ofMerger
- 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
<VR> KTable<Windowed<K>,VR> aggregate(Initializer<VR> initializer, Aggregator<? super K,? super V,VR> aggregator, Merger<? super K,VR> sessionMerger, Materialized<K,VR,SessionStore<org.apache.kafka.common.utils.Bytes,byte[]>> materialized)
Aggregate the values of records in this stream by the grouped key and definedSessionWindows
. Records withnull
key or value are ignored. The result is written into a localSessionStore
(which is basically an ever-updating materialized view) that can be queried using the name provided withMaterialized
. Aggregating is a generalization ofcombining via reduce(...)
as it, for example, allows the result to have a different type than the input values.The specified
Initializer
is applied once per session directly before the first input record is processed to provide an initial intermediate aggregation result that is used to process the first record. The specifiedAggregator
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 theInitializer
) and the record's value. * The specifiedMerger
is used to merge 2 existing sessions into one, i.e., when the windows overlap, they are merged into a single session and the old sessions are discarded. Thus,aggregate(Initializer, Aggregator, Merger)
can be used to compute aggregate functions like count (c.f.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 theconfiguration
parameters forcache size
, andcommit intervall
KafkaStreams streams = ... // some windowed aggregation on value type double Sting queryableStoreName = ... // the queryableStoreName should be the name of the store as defined by the Materialized instance ReadOnlySessionStore<String, Long> sessionStore = streams.store(queryableStoreName, QueryableStoreTypes.<String, Long>sessionStore()); String key = "some-key"; KeyValueIterator<Windowed<String>, Long> aggForKeyForSession = localWindowStore.fetch(key); // key must be local (application state is shared over all running Kafka Streams instances)
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 parameterAPPLICATION_ID_CONFIG
, "storeName" is the provide store name defined inMaterialized
, and "-changelog" is a fixed suffix. You can retrieve all generated internal topic names viaTopology.describe()
.- Type Parameters:
VR
- the value type of the resultingKTable
- Parameters:
initializer
- the instance ofInitializer
aggregator
- the instance ofAggregator
sessionMerger
- the instance ofMerger
materialized
- an instance ofMaterialized
used to materialize a state store. Cannot benull
- 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
KTable<Windowed<K>,V> reduce(Reducer<V> reducer)
Combine values of this stream by the grouped key intoSessionWindows
. Records withnull
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.aggregate(Initializer, Aggregator, Merger)
). The result is written into a localSessionStore
(which is basically an ever-updating materialized view).The specified
Reducer
is applied for each input record and computes a new aggregate using the current aggregate and the record's value. If there is no current aggregate theReducer
is not applied and the new aggregate will be the record's value as-is. Thus,reduce(Reducer)
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 forcache size
, andcommit intervall
.
-
reduce
KTable<Windowed<K>,V> reduce(Reducer<V> reducer, Materialized<K,V,SessionStore<org.apache.kafka.common.utils.Bytes,byte[]>> materializedAs)
Combine values of this stream by the grouped key intoSessionWindows
. Records withnull
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.aggregate(Initializer, Aggregator, Merger)
). The result is written into a localSessionStore
(which is basically an ever-updating materialized view) provided by the givenMaterialized
instance.The specified
Reducer
is applied for each input record 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; } }
If there is no current aggregate the
Reducer
is not applied and the new aggregate will be the record's value as-is. Thus,reduce(Reducer, Materialized)
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 theconfiguration
parameters forcache size
, andcommit intervall
To query the local windowed
KeyValueStore
it must be obtained viaKafkaStreams#store(...)
.
For non-local keys, a custom RPC mechanism must be implemented usingKafkaStreams streams = ... // compute sum Sting queryableStoreName = ... // the queryableStoreName should be the name of the store as defined by the Materialized instance ReadOnlySessionStore<String,Long> localWindowStore = streams.store(queryableStoreName, QueryableStoreTypes.<String, Long>ReadOnlySessionStore<String, Long>); String key = "some-key"; KeyValueIterator<Windowed<String>, Long> sumForKeyForWindows = localWindowStore.fetch(key); // key must be local (application state is shared over all running Kafka Streams instances)
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 parameterAPPLICATION_ID_CONFIG
, "storeName" is the provide store name defined inMaterialized
, and "-changelog" is a fixed suffix. You can retrieve all generated internal topic names viaTopology.describe()
.- Parameters:
reducer
- aReducer
that computes a new aggregate result. Cannot benull
.materializedAs
- an instance ofMaterialized
used to materialize a state store. Cannot benull
- 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
-
-