Kafka streams: how to compare current and previous values on stream

[Davide Pozza]

Hi all

I need to create a stream processor which should publish an alert (message) on an output topic only when the values from the input stream change.

The input stream publishes a message like:

<CLIENT_ID> => <VALUE>

When the <VALUE> for a specific <CLIENT_ID> changes (I mean the current one is different from the previous one), my stream processor should publish on the output topic the CLIENT_ID for which the value changed:

<NEW_KEY> => <CLIENT_ID>

I suppose I should store on a KTable (windowed?) the "previous" values in order to compare them (through a join) with the new ones coming from the input stream, but I'm unable to figure out...

Any suggestions?

Thank you in advance!

Davide

[Matthias J. Sax]

I think, it would be the best to user Processor API for this. Create a
KeyValueStore and assign it to your processor. Within the processor, for
each record you do a lookup in the KeyValue store to compare with the
previous record of the client. If the value did change, emit and update
key-value store.

One thing you need to keep in mind, is out-of-order data. Thus, you
should also check the offset of the currently processed record via the
provided context from init().


-Matthias

> --
> You received this message because you are subscribed to the Google
> Groups "Confluent Platform" group.
> To unsubscribe from this group and stop receiving emails from it, send
> an email to confluent-platf...@googlegroups.com
> <mailto:confluent-platf...@googlegroups.com>.
> To post to this group, send email to confluent...@googlegroups.com
> <mailto:confluent...@googlegroups.com>.
> To view this discussion on the web visit
> https://groups.google.com/d/msgid/confluent-platform/8019f389-df22-47e6-8084-80f0fa60b573%40googlegroups.com
> <https://groups.google.com/d/msgid/confluent-platform/8019f389-df22-47e6-8084-80f0fa60b573%40googlegroups.com?utm_medium=email&utm_source=footer>.
> For more options, visit https://groups.google.com/d/optout.

[Guozhang Wang]

We may support it in the future at the DSL layer as a windowed-table self-join with shifted windows, but it is not yet supported now so I think the Processor API as Matthias mentioned is the way to go for now.

Guozhang

> an email to confluent-platform+unsub...@googlegroups.com
> <mailto:confluent-platform+unsub...@googlegroups.com>.

[Davide Pozza]

Thank you!

[Maciej Lopacinski]

Hi guys,

My use case is similar, I was wondering if since 2016 there was any change in the API that makes the mentioned implementation possible.

@Davide - did you have a chance to implement the KeyValueStore solution?