Hey there, I am pulling data in from network interfaces and trying to figure out a way to get the delta of the bytes sent/received between the records. The data is sent by the devices once a minute with a row for each network interface in the device. I am not sure if this is something that can be calculated on ingestion (or if it should be if it can be) or if it needs to be done via a query. The data should look like the following with each row being a separate message pulled in from Kafka:

__time	                    Device ID	interface-name	    bytes-sent	bytes-received	Delta Bytes Sent	Delta Bytes Rcvd
2024-05-24T18:05:23.120Z	<device-1>	<dev1-interface-A>	130978575677	8420959997	889758	60650
2024-05-24T18:05:23.120Z	<device-2>	<dev2-interface-B>	2945859418	3074108147129	5310	6747698
2024-05-24T18:04:23.052Z	<device-1>	<dev1-interface-A>	130977685919	8420899347	356143	28719
2024-05-24T18:04:23.052Z	<device-2>	<dev2-interface-B>	2945854108	3074101399431	5246	6830232
2024-05-24T18:03:23.065Z	<device-1>	<dev1-interface-A>	130977329776	8420870628	356465	31700
2024-05-24T18:03:23.065Z	<device-2>	<dev2-interface-B>	2945848862	3074094569199	5182	6676525
2024-05-24T18:02:23.088Z	<device-1>	<dev1-interface-A>	130976973311	8420838928	357472	29460
2024-05-24T18:02:23.088Z	<device-2>	<dev2-interface-B>	2945843680	3074087892674	5310	6774756
2024-05-24T18:01:23.062Z	<device-1>	<dev1-interface-A>	130976615839	8420809468		
2024-05-24T18:01:23.062Z	<device-2>	<dev2-interface-B>	2945838370	3074081117918

The first 5 columns are part of the data set we consume and the last 2 columns are the ones I want to calculate on ingestion or via a query. We are running Druid 28.0.1 and the data is being pulled in as JSON messages on a Kafka topic Any help will be appreciated :)

This can be done at ingestion time. You would need to add a tranformation that contains the expression to calculate the delta.

Thanks for this although I am still having issues trying to figure out how to do this with expressions in the transform spec. Using the above table with the first row as #1, then 2, etc. the formula is (row1.bytes-sent - row3.bytes-sent) although more specifically, that breaks down to having

previous_row.__time == (current_row.time - 1 minute) && 
previous_row.deviceID == current_row.deviceID && 
previous_row.interface-name == current_row.interface-name

but I do not see a way to query previous rows using only expressions

Streaming ingestion cannot read previous rows (since rows can be distributed among multiple tasks). You would need to either do the calculation at query time, or do a batch reprocessing of the data after it has been ingested.

Thanks, I thought I was missing something for ingestion (and trying to figure out how it would work for distributed ingestion). I will look into doing it as a query or maybe as a preprocessing step using something like Flink

You may want to look up some of the case studies with NTT, they are using Druid and are doing similar calculations. I believe they are using realtime aggregation to sum the bytes sent and bytes received, so the delta is calculated over aggregated data instead of each row.