Spark withWatermark, only store values before and after a gap

Viewed 125

I've got data comming in over Kafka to a Spark Structured Streaming application. To simplify, each message on kafka contains a device-id, a datetime and a value. The purpose of the streaming application is to calculate the difference between all values for each device.

I.e. if the input is

Device-ID Datetime Value
1 20210922-15:15 21
Device-ID Datetime Value
1 20210922-15:16 24
Device-ID Datetime Value
1 20210922-15:17 26

I would like the output to be

Device-ID Datetime Value
1 20210922-15:16 3
1 20210922-15:17 2

To solve this, and to handle that messages can come in late (up to 10 days) I'm using withWatermark on the Dateteime-column, with 10 days. However this leads to a huge memory usage if I have many devices since spark will store all the values for 10 days for all devices in memory.

In practice however, I do not have the need to store e.g., the value for 15:16 for Device X, if I already have retreived the values for 15:15 and 15:17.

So, instead of storing something like this in memory (due to withWatermark)

Device-ID Datetime Value
1 20210922-15:15
1 20210922-15:16
1 20210922-15:17
1 20210922-15:18
2 20210922-14:15
2 20210922-14:16
2 20210922-14:17
2 20210922-14:18

I would only need this

Device-ID Datetime Value
1 20210922-15:15
1 20210922-15:18
2 20210922-14:15
2 20210922-14:18

Is this doable?

0 Answers
Related