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?