KStream value filtering based on a KTable

user_3369840 0
Viewed 58

I need to filter a KStream based on the values of a KTable. This is similar to the SQL IN clause. I went through the documentation for KStream filter method but i could not find a way to filter based on a KTable.

PS. I do not want to use kSQLDb

apache-kafka-streams apache-kafka java
1 Answers

In KAFKA Filter is a Stateless Transformation Operation and it is pretty simple to implement it in both KStream and KTable.

Basic Definition

Filter KStream → KStream and KTable → KTable

Evaluates a boolean function for each element and retains those for which the function returns true.

KStream<String, Long> stream = ...;

// A filter that selects (keeps) only positive numbers
// Java 8+ example, using lambda expressions
KStream<String, Long> onlyPositives = stream.filter((key, value) -> value > 0);

You can refer this simple example for the usage of filter to chuck out the odd numbers from an input stream.

https://github.com/confluentinc/kafka-streams-examples/blob/7.1.1-post/src/main/java/io/confluent/examples/streams/SumLambdaExample.java

user_105840610
Related