Hello, can't understand how to do next: I've got few streams from few Kafka topics (if it possible to do other way and easier I could make one topic or any other modifications) with sensors measurements into JSON messages:

topic1: {'data': {'temp':25.2, 'sensore_name': 'T1', 'timestamp': 123123131}, 'compare_with': 'T2'}
topic2: {'data': {'temp':28, 'sensore_name': 'T2', 'timestamp': 53543543}, 'compare_with': 'T1'}
topic3: {'data': {'temp':32, 'sensore_name': 'T3', 'timestamp': 6757575}, 'compare_with': 'T2'}
topic4: {'data': {'temp':12, 'sensore_name': 'T3', 'timestamp': 67856222}, 'compare_with': 'T1'}

I need to compare T1.data.temp - T2.data.temp (I need to compare it with EXACTLY last measurement of other sensor (Shown in compare_with), because measurements could come with different frequency: T1 1 message per sec, T2 1 message per 5 sec., T3 3 message per sec.) calculate AVG from this difference in 1 houre window, and if this difference more than AVG, then make Alarm to somewhere... Don't understand how to do it? Flink 1.17.1 Java 17

Hi You can use reduce operator, by that you can access your last log and perform whatever operation you want.

Sorry, but I can't understand how reduce can helps me in my case? How to calculate difference of temps from 2 streams? Now i did like this:

DataStream<MergedTemps> comparisonStream = T1_stream
	.join(T2_stream)
	.where(T1 -> T1.getCompare_with())
	.equalTo(T2 -> T2.getSensore_name())
	.window(TumblingEventTimeWindows.of(Time.seconds(1)))
	.apply((JoinFunction<Sensor, Sensor, MergedTemps>) (T1, T2) -> {

		Tuple2<Double, Double> spread;
		
		double a = T1.getData().getTemp();
		double b = T2.getData().getTemp();
		spread = new Tuple2<>(a, b);
		
		return new MergedTemps(T1.getTimestamp(), T1.getSensore_name(), spread);
	});

But if I understand right: it merge mesurements not 1 last to 1 last...

Can you elaborate you want difference between last 2 element if yes then you can do with the Reduce by appending in existing operator If no you need to create windowing

I need difference between last 2 elements IN TWO DIFFEREND STREAMS! Last lelement from stream T1 - last element from stream T2! Stream<Double>= T1.data.temp - T2.data.temp That's why I'm not sure about reduce function, as I understand it works with one stream.

Okay, I understand