- html - 出于某种原因,IE8 对我的 Sass 文件中继承的 html5 CSS 不友好?
- JMeter 在响应断言中使用 span 标签的问题
- html - 在 :hover and :active? 上具有不同效果的 CSS 动画
- html - 相对于居中的 html 内容固定的 CSS 重复背景?
我正在 Java 中实现一个 Flink 程序,以使用 MapStateDescriptor
处理状态。我的实现基于此 source 。由于某种原因,MapState
保留了以前的值,我无法计算出我想要的平均值。当我调试时,averageTemp
始终为空,并且我从未在其中找到任何 key 。我的实现中缺少什么?
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 个线程,并且在开始计算平均值之前它会覆盖我的 map 的值至少 3 次。
最佳答案
map 功能中的状态以每个键为基础。因此,当调用 map 函数并获取 map 状态时,它将针对正在处理的 MqttTemperature
记录中的任何 id。
鉴于您想要每个房间的平均温度,我的处理方式如下:
TemperatureKeySelector
以返回 room-A
、room-B
或 room-C
id 字段。AverageTempMapper
中,有两个 ValueState
变量 - 一个是温度总和(Double),另一个是计数。当调用 map()
方法时,如果这两个 ValueState
变量中的任何一个为 null,请将其初始化为 0,然后求和/递增。关于java - 为什么 Flink 中的 MapState 变量没有保留以前的值?,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/54478033/
这个问题在这里已经有了答案: 关闭 10 年前。 Possible Duplicate: How to nest OR statements in JavaScript? 有没有办法做到这一点:
在 JavaScript 中有没有办法让一个变量总是等于一个变量?喜欢var1 = var2但是当var2更新,也是var1 . 例子 var var1 = document.getElementBy
我正在努力理解这代表什么 var1 = var2 == var3 我的猜测是这等同于: if (var2 == var3): var1 = var2 最佳答案 赋值 var1 = var2
这个问题已经有答案了: What does the PHP error message "Notice: Use of undefined constant" mean? (2 个回答) 已关闭 8
我在临时表中有几条记录,我想从每条记录中获取一个值并将其添加到一个变量中,例如 color | caption -------------------------------- re
如何将字符串转为变量(字符串变量--> $variable)? 或者用逗号分隔的变量列表然后转换为实际变量。 我有 2 个文件: 列名文件 行文件 我需要根据字符串匹配行文件中的整行,并根据列名文件命
我有一个我无法解决的基本 php 问题,我也想了解为什么! $upperValueCB = 10; $passNodeMatrixSource = 'CB'; $topValue= '$uppe
这可能吗? php $variable = $variable1 || $variable2? 如果 $variable1 为空则使用 $variable2 是否存在类似的东西? 最佳答案 PHP 5
在 Perl 5.20 中,for 循环似乎能够修改模块作用域的变量,但不能修改父作用域中的词法变量。 #!/usr/bin/env perl use strict; use warnings; ou
为什么这不起作用: var variable; variable = variable.concat(variable2); $('#lunk').append(variable) 我无法弄清楚这一点
根据我的理解,在32位机器上,指针的sizeof是32位(4字节),而在64位机器上,它是8字节。无论它们指向什么数据类型,它们都有固定的大小。我的计算机在 64 位上运行,但是当我打印包含 * 的大
例如: int a = 10; a += 1.5; 这运行得很完美,但是 a = a+1.5; 此作业表示类型不匹配:无法从 double 转换为 int。所以我的问题是:+= 运算符 和= 运算符
您好,我写了这个 MySQL 存储过程,但我一直收到这个语法错误 #1064 - You have an error in your SQL syntax; check the manual that
我试图在我的场景中显示特定的奖牌,这取决于你的高分是基于关卡的目标。 // Get Medal Colour if levelHighscore goalScore { sc
我必须维护相当古老的 Visual C++ 源代码的大型代码库。我发现代码如下: bIsOk = !!m_ptr->isOpen(some Parameters) bIsOk的数据类型是bool,is
我有一个从 MySQL 数据库中提取的动态产品列表。在 list 上有一个立即联系 按钮,我正在使用一个 jquery Modal 脚本,它会弹出一个表单。 我的问题是尝试将产品信息变量传递给该弹出窗
这个问题在这里已经有了答案: 关闭 10 年前。 Possible Duplicate: What is the difference between (type)value and type(va
jQuery Core Style Guidelines建议两种不同的方法来检查变量是否已定义。 全局变量:typeof variable === "undefined" 局部变量:variable
这个问题已经有答案了: 已关闭11 年前。 Possible Duplicate: “Variable” Variables in Javascript? 我想肯定有一种方法可以在 JavaScrip
在语句中使用多重赋值有什么优点或缺点吗?在简单的例子中 var1 = var2 = true; 赋值是从右到左的(我相信 C# 中的所有赋值都是如此,而且可能是 Java,尽管我没有检查后者)。但是,
我是一名优秀的程序员,十分优秀!