java - 为什么我在 Flink 中的 MapState 变量没有保留以前的值?
问题描述
我正在用 Java 实现一个 Flink 程序来处理使用MapStateDescriptor
. 我基于这个源实现。出于某种原因,MapState
它保留了以前的值,我无法计算出我想要的平均值。当我调试时,averageTemp
它总是空的,我从来没有在里面找到任何钥匙。我的实施缺少什么?
import java.util.Map;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.state.MapState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.sense.flink.mqtt.MqttTemperature;
import org.sense.flink.mqtt.TemperatureMqttConsumer;
public class SensorsMultipleReadingMqttEdgentQEP {
public SensorsMultipleReadingMqttEdgentQEP() throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setStreamTimeCharacteristic(TimeCharacteristic.IngestionTime);
DataStream<MqttTemperature> temperatureStream01 = env.addSource(new TemperatureMqttConsumer("topic-edgent-01"));
DataStream<MqttTemperature> temperatureStream02 = env.addSource(new TemperatureMqttConsumer("topic-edgent-02"));
DataStream<MqttTemperature> temperatureStream03 = env.addSource(new TemperatureMqttConsumer("topic-edgent-03"));
DataStream<MqttTemperature> temperatureStreams = temperatureStream01.union(temperatureStream02)
.union(temperatureStream03);
DataStream<Tuple2<String, Double>> average = temperatureStreams.keyBy(new TemperatureKeySelector())
.map(new AverageTempMapper());
average.print();
env.execute("SensorsMultipleReadingMqttEdgentQEP");
}
public static class TemperatureKeySelector implements KeySelector<MqttTemperature, Integer> {
private static final long serialVersionUID = 5905504239899133953L;
@Override
public Integer getKey(MqttTemperature value) throws Exception {
return value.getId();
}
}
public static class AverageTempMapper extends RichMapFunction<MqttTemperature, Tuple2<String, Double>> {
private static final long serialVersionUID = -5489672634096634902L;
private MapState<String, Double> averageTemp;
@Override
public void open(Configuration parameters) throws Exception {
averageTemp = getRuntimeContext()
.getMapState(new MapStateDescriptor<>("average-temperature", String.class, Double.class));
}
@Override
public Tuple2<String, Double> map(MqttTemperature value) throws Exception {
String key = "no-room";
Double temp = value.getTemp();
if (value.getId().equals(1) || value.getId().equals(2) || value.getId().equals(3)) {
key = "room-A";
} else if (value.getId().equals(4) || value.getId().equals(5) || value.getId().equals(6)) {
key = "room-B";
} else if (value.getId().equals(7) || value.getId().equals(8) || value.getId().equals(9)) {
key = "room-C";
}
// NEVER ITERATES ON THE averageTemp
for (Map.Entry<String, Double> entry: averageTemp.entries()) {
System.out.println(entry.getKey() + " - " + entry.getValue());
}
System.out.println("value: " + value);
if (averageTemp.contains(key)) { // NEVER CONTAINS A KEY
System.out.println("yes: " + key);
temp = (averageTemp.get(key) + value.getTemp()) / 2;
} else {
averageTemp.put(key, temp);
}
return Tuple2.of(key, temp);
}
}
}
**编辑:**好的。我误解了这个问题。该代码将先前的状态保存在 MapState 上。我错了,因为我正在调试代码。但实际上我遇到的问题是它启动了超过 1 个线程,并且在开始计算平均值之前它至少覆盖了我的地图值 3 次。
解决方案
map 函数中的状态是基于每个键的。因此,当您的地图函数被调用并获得地图状态时,它将针对MqttTemperature
正在处理的记录中的任何 id。
鉴于您想要每个房间的平均温度,我将处理的方式如下:
- 将 更改
TemperatureKeySelector
为 returnroom-A
,room-B
或room-C
基于 id 字段。 - 在 中
AverageTempMapper
,有两个ValueState
变量 - 一个是温度的总和(一个 Double),另一个是计数。当您的map()
方法被调用时,如果这两个ValueState
变量中的任何一个为空,则将其初始化为 0,然后求和/递增。
推荐阅读
- python - 无法在python中打印描述文本
- c# - Switch 语句没有按预期工作,即使语法可能完全正确
- android - 从微调器中选择并单击按钮以打开新活动
- python - Kivy:如何正确绑定下拉菜单?
- javascript - 从 API 获取数据并将其动态添加到表中
- shopware - 如何在电子邮件模板中使用订单自定义字段值?
- google-chrome-extension - 如何使用 chrome.storage api 存储来自我的 chrome 扩展的 background.js 文件中定义的这 2 个事件监听器的数据(url、时间戳)
- spring-boot - Spring Boot Rest Api - 多个 @RequestPart 不适用于 Feign 客户端
- python - 小部件结构
- pentaho - 循环仅激活一次 Pentaho