How to log filtered data in Kafka Stream?

Cr4zyTun4 0
Viewed 39

I hava a kafka Stream and I perform a filter operation. I would like to log those records that get filtered out.

streamsBuilder.stream(topicName)
.filterNot( (k,v) -> v.getExampleProperty() == null)
...

I would like to log those record with null ExampleProperty.
I would like to use something like the peek() function but with some predicate. How to do that?

apache-kafka-streams apache-kafka
2 Answers

The way you use it, you can write it directly behind the filter.
"Filter creates a new KStream that consists of all records of this stream which satisfy the given predicate." The filter contains the same predicate as the peek:

streamsBuilder.stream(topicName)
.filter( (k,v) -> v.getExampleProperty() == null)
.peek( (k,v) -> {System.out.printf("key:" + k + "value" + v)})

when you want peek the filtered out records you could use the complementary filter in parallel:

.stream(topicName)
.filterNot( (k,v) -> v.getExampleProperty() == null)

If you do not want to filter twice the same stream you could use something like split(branch) instead of the filter(more on this here: https://stackoverflow.com/a/40921376/5528518)

WnaJ0

EDIT:

This is how I thought the problem could be solved:

.peek( (k,v) -> { 
    if (v.getExampleProperty() == null){ 
       System.out.printf("key: " + k.toString()); 
       System.out.printf("value: " + v.toString());
    }
})

However as pointed out by @OneCricketeer in the comments this approach could process unnecessary data. It is better to use Filter and FilterNot followed by a peek for such a use case.

.stream(topicName)
.filter( (k,v) -> v.getExampleProperty() == null)
# ... your other code

# the filtered out print
.stream(topicName)
.filterNot( (k,v) -> v.getExampleProperty() == null)
.peek( (k,v) -> { 
   System.out.printf("key: " + k.toString()); 
   System.out.printf("value: " + v.toString());
}
Cr4zyTun40
Related