org.apache.kafka.streams.kstream

Interface KTable<K,V>

Type Parameters:

K - Type of primary keys

V - Type of value changes

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

KTable is an abstraction of a changelog stream from a primary-keyed table. Each record in this stream is an update on the primary-keyed table with the record key as the primary key.

A KTable is either defined from one or multiple Kafka topics that are consumed message by message or the result of a KTable transformation. An aggregation of a KStream also yields a KTable.

A KTable can be transformed record by record, joined with another KTable or KStream, or can be re-partitioned and aggregated into a new KTable.

See Also:

KStream

Method Summary

Methods

Modifier and Type	Method and Description
KTable<K,V>	filter(Predicate<K,V> predicate) Create a new instance of KTable that consists of all elements of this stream which satisfy a predicate.
KTable<K,V>	filterNot(Predicate<K,V> predicate) Create a new instance of KTable that consists all elements of this stream which do not satisfy a predicate.
void	foreach(ForeachAction<K,V> action) Perform an action on each element of KTable.
String	getStoreName() Get the name of the local state store used for materializing this KTable
<K1,V1> KGroupedTable<K1,V1>	groupBy(KeyValueMapper<K,V,KeyValue<K1,V1>> selector) Group the records of this KTable using the provided KeyValueMapper and default serializers and deserializers.
<K1,V1> KGroupedTable<K1,V1>	groupBy(KeyValueMapper<K,V,KeyValue<K1,V1>> selector, Serde<K1> keySerde, Serde<V1> valueSerde) Group the records of this KTable using the provided KeyValueMapper.
<V1,R> KTable<K,R>	join(KTable<K,V1> other, ValueJoiner<V,V1,R> joiner) Combine values of this stream with another KTable stream's elements of the same key using Inner Join.
<V1,R> KTable<K,R>	leftJoin(KTable<K,V1> other, ValueJoiner<V,V1,R> joiner) Combine values of this stream with another KTable stream's elements of the same key using Left Join.
<V1> KTable<K,V1>	mapValues(ValueMapper<V,V1> mapper) Create a new instance of KTable by transforming the value of each element in this stream into a new value in the new stream.
<V1,R> KTable<K,R>	outerJoin(KTable<K,V1> other, ValueJoiner<V,V1,R> joiner) Combine values of this stream with another KTable stream's elements of the same key using Outer Join.
void	print() Print the elements of this stream to System.out.
void	print(Serde<K> keySerde, Serde<V> valSerde) Print the elements of this stream to System.out
void	print(Serde<K> keySerde, Serde<V> valSerde, String streamName) Print the elements of this stream to System.out
void	print(String streamName) Print the elements of this stream to System.out.
KTable<K,V>	through(Serde<K> keySerde, Serde<V> valSerde, StreamPartitioner<K,V> partitioner, String topic, String storeName) Materialize this stream to a topic, also creates a new instance of KTable from the topic using a customizable StreamPartitioner to determine the distribution of records to partitions.
KTable<K,V>	through(Serde<K> keySerde, Serde<V> valSerde, String topic, String storeName) Materialize this stream to a topic, also creates a new instance of KTable from the topic.
KTable<K,V>	through(StreamPartitioner<K,V> partitioner, String topic, String storeName) Materialize this stream to a topic, also creates a new instance of KTable from the topic using default serializers and deserializers and a customizable StreamPartitioner to determine the distribution of records to partitions.
KTable<K,V>	through(String topic, String storeName) Materialize this stream to a topic, also creates a new instance of KTable 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.
KStream<K,V>	toStream() Convert this stream to a new instance of KStream.
<K1> KStream<K1,V>	toStream(KeyValueMapper<K,V,K1> mapper) Convert this stream to a new instance of KStream using the given KeyValueMapper to select the new key.
void	writeAsText(String filePath) Write the elements of this stream to a file at the given path using default serializers and deserializers.
void	writeAsText(String filePath, Serde<K> keySerde, Serde<V> valSerde) Write the elements of this stream to a file at the given path.
void	writeAsText(String filePath, String streamName) Write the elements of this stream to a file at the given path.
void	writeAsText(String filePath, String streamName, Serde<K> keySerde, Serde<V> valSerde)

Method Detail

filter

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

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

Parameters:

predicate - the instance of Predicate

Returns:

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

filterNot

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

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

Parameters:

predicate - the instance of Predicate

Returns:

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

mapValues

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

Create a new instance of KTable 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 KTable that contains records with unmodified keys and new values of different type

print

void print()

Print the elements of this stream to System.out. This function will use the generated name of the parent processor node to label the key/value pairs printed out to the console. 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(String streamName)

Print the elements of this stream to System.out. This function will use the given name to label the key/value printed out to the console.

Parameters:

streamName - the name used to label the key/value pairs printed out to the console 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.

print

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

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

streamName - the name used to label the key/value pairs printed out to the console 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 using default serializers and deserializers.

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,
               String streamName)

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

Parameters:

filePath - name of file to write to

streamName - the name used to label the key/value pairs printed out to the console 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)

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

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.

writeAsText

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

Parameters:

filePath - name of file to write to

streamName - the name used to label the key/value pairs printed out to the console

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.

through

KTable<K,V> through(String topic,
                  String storeName)

Materialize this stream to a topic, also creates a new instance of KTable from the topic using default serializers and deserializers and producer's DefaultPartitioner. This is equivalent to calling to(String) and KStreamBuilder.table(String, String). The resulting KTable will be materialized in a local state store with the given store name. Also a changelog topic named "${applicationId}-${storeName}-changelog" will be automatically created in Kafka for failure recovery, where "applicationID" is specified by the user in StreamsConfig.

Parameters:

topic - the topic name

storeName - the state store name used for this KTable

Returns:

a new KTable that contains the exact same records as this KTable

through

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

Materialize this stream to a topic, also creates a new instance of KTable 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(String) and KStreamBuilder.table(String, String). The resulting KTable will be materialized in a local state store with the given store name. Also a changelog topic named "${applicationId}-${storeName}-changelog" will be automatically created in Kafka for failure recovery, where "applicationID" is specified by the user in StreamsConfig.

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

storeName - the state store name used for this KTable

Returns:

a new KTable that contains the exact same records as this KTable

through

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

Materialize this stream to a topic, also creates a new instance of KTable 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.table(Serde, Serde, String, String). The resulting KTable will be materialized in a local state store with the given store name. Also a changelog topic named "${applicationId}-${storeName}-changelog" will be automatically created in Kafka for failure recovery, where "applicationID" is specified by the user in StreamsConfig.

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

storeName - the state store name used for this KTable

Returns:

a new KTable that contains the exact same records as this KTable

through

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

Materialize this stream to a topic, also creates a new instance of KTable 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.table(Serde, Serde, String, String). The resulting KTable will be materialized in a local state store with the given store name. Also a changelog topic named "${applicationId}-${storeName}-changelog" will be automatically created in Kafka for failure recovery, where "applicationID" is specified by the user in StreamsConfig.

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

storeName - the state store name used for this KTable

Returns:

a new KTable that contains the exact same records as this KTable

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

toStream

KStream<K,V> toStream()

Convert this stream to a new instance of KStream.

Returns:

a KStream that contains the same records as this KTable; the records are no longer treated as updates on a primary-keyed table, but rather as normal key-value pairs in a record stream

toStream

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

Convert this stream to a new instance of KStream using the given KeyValueMapper to select the new key.

Type Parameters:

K1 - the new key type

Parameters:

mapper - @param mapper the instance of KeyValueMapper

Returns:

a KStream that contains the transformed records from this KTable; the records are no longer treated as updates on a primary-keyed table, but rather as normal key-value pairs in a record stream

join

<V1,R> KTable<K,R> join(KTable<K,V1> other,
                      ValueJoiner<V,V1,R> joiner)

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

Type Parameters:

V1 - the value type of the other stream

R - the value type of the new stream

Parameters:

other - the instance of KTable joined with this stream

joiner - the instance of ValueJoiner

Returns:

a KTable that contains join-records for each key and values computed by the given ValueJoiner, one for each matched record-pair with the same key

outerJoin

<V1,R> KTable<K,R> outerJoin(KTable<K,V1> other,
                           ValueJoiner<V,V1,R> joiner)

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

Type Parameters:

V1 - the value type of the other stream

R - the value type of the new stream

Parameters:

other - the instance of KTable joined with this stream

joiner - the instance of ValueJoiner

Returns:

a KTable that contains join-records for each key and values computed by the given ValueJoiner, one for each matched record-pair with the same key

leftJoin

<V1,R> KTable<K,R> leftJoin(KTable<K,V1> other,
                          ValueJoiner<V,V1,R> joiner)

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

Type Parameters:

V1 - the value type of the other stream

R - the value type of the new stream

Parameters:

other - the instance of KTable joined with this stream

joiner - the instance of ValueJoiner

Returns:

a KTable that contains join-records for each key and values computed by the given ValueJoiner, one for each matched record-pair with the same key

groupBy

<K1,V1> KGroupedTable<K1,V1> groupBy(KeyValueMapper<K,V,KeyValue<K1,V1>> selector,
                                   Serde<K1> keySerde,
                                   Serde<V1> valueSerde)

Group the records of this KTable using the provided KeyValueMapper.

Type Parameters:

K1 - the key type of the KGroupedTable

V1 - the value type of the KGroupedTable

Parameters:

selector - select the grouping key and value to be aggregated

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

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

Returns:

a KGroupedTable that contains the re-partitioned records of this KTable

groupBy

<K1,V1> KGroupedTable<K1,V1> groupBy(KeyValueMapper<K,V,KeyValue<K1,V1>> selector)

Group the records of this KTable using the provided KeyValueMapper and default serializers and deserializers.

Type Parameters:

K1 - the key type of the KGroupedTable

V1 - the value type of the KGroupedTable

Parameters:

selector - select the grouping key and value to be aggregated

Returns:

a KGroupedTable that contains the re-partitioned records of this KTable

foreach

void foreach(ForeachAction<K,V> action)

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

Parameters:

action - an action to perform on each element

getStoreName

String getStoreName()

Get the name of the local state store used for materializing this KTable

Returns:

the underlying state store name, or null if KTable does not have one