3 Kafka Streams -> Stateless Operations of KStreams and KTables

Definition

KStreams -> like logs, which is insert only and it is infinity

KTables

• All upserts on non null values
• Deletes on null values
• Similar to a table
• Parallel with log compacted topics

Introduce KTable:

Stateless vs Stateful

Stateless means that the result of a transformation only depends on the data-point you process

example (y = 2x):     1 -> 2 , 300 -> 600. In this case you only focus on the current state of data/operation.

Stateful means that the result of a transformation also depends on external information - the state

example: hello -> , hello -> 2. the second hello is 2, because the calculation based on the first one.

Common Kafka Streams Operations

  

    

 

 

Example for Brach:

  

Read and write from / to Topic:

   

Set Log Compaction

kafka-topics --zookeeper localhost:2181 --create --topics test --partitions 1 --replication-factor 3 \

　　　　　  --config clean.policy=compact \

　　　　　  --config min.cleanable.dirty.ratio=0.5 \

　　　　　  --config segment.ms=10000 

 Example below shows

1. create a topic with log-compaction

2. produce (background) and consume data

3. consume data again after log-compaction happened.

 

 Transform between KTable and KStream

1. Table to Stream

Sometimes it is helpful to transform a KTable to a KStream in order to keep a changelog of all the changes to the KTable!

This can be easily achieved in one line of code:

KTable<byte[], String> table = ... ;
// also a variant of 'toStream' exists that allows you to select a new key for the resulting stream
KStream<byte[], String> stream = table.toStream();

2. Stream to Table

There are two way:

• Perform a chain of groupByKey() and followed with a aggregation like count, aggregate or reduce

KTable<String, Long> table = stream.groupByKey().count();

•  Write back to Kafka and read as KTable (it will create a intermediate topic and also some network traffic)

// write to Kafka
stream.to("my-topic");

// read from Kafka as a table
KTable<String, String> table = builder.table("my-topic");

With this operation you will have all of advatages of KTable, which means it will receive all of update state and delete ( if value is null ) information of a Stream.