我正在Flink中接收一个数据流。对于这个数据的每个"实例"，我都有一个时间戳。我可以检测我从中获取数据的机器是"生产"还是"不生产"，这是通过位于它自己的静态类中的自定义平面图函数完成的。
我想计算一下这台机器生产/不生产的时间。我目前的方法是在两个简单列表中收集生产和非生产时间戳。对于数据的每个"实例"，我通过从最早的时间戳中减去最新的时间戳来计算当前的生产/非生产持续时间。不过，这给了我不正确的结果。当生产状态从生产变为非生产时，我清除生产的时间戳列表，反之亦然，这样，如果生产再次开始，持续时间从零开始。
我查看了我收集的两个列表中各自的时间戳，发现了一些我不理解的东西。我的假设是，只要机器"生产"，生产时间戳列表中的第一个时间戳就会保持不变，而每个新的数据实例都会向列表中添加新的时间戳。令人惊讶的是，这个假设是错误的，因为我在列表中得到了看似随机的时间戳。不过，它们的顺序仍然正确。
这是我的平面图功能代码：
public static class ImaginePaperDataConverterRich extends RichFlatMapFunction<ImaginePaperData, String> {
private static final long serialVersionUID = 4736981447434827392L;
private transient ValueState<ProductionState> stateOfProduction;
SimpleDateFormat dateFormat = new SimpleDateFormat("dd.MM.yyyy HH:mm:ss.SS");
DateFormat timeDiffFormat = new SimpleDateFormat("dd HH:mm:ss.SS");
String timeDiffString = "00 00:00:00.000";
List<String> productionTimestamps = new ArrayList<>();
List<String> nonProductionTimestamps = new ArrayList<>();
public String calcProductionTime(List<String> timestamps) {
if (!timestamps.isEmpty()) {
try {
Date firstDate = dateFormat.parse(timestamps.get(0));
Date lastDate = dateFormat.parse(timestamps.get(timestamps.size()-1));
long timeDiff = lastDate.getTime() - firstDate.getTime();
if (timeDiff < 0) {
System.out.println("Something weird happened. Maybe EOF.");
return timeDiffString;
}
timeDiffString = String.format("%02d %02d:%02d:%02d.%02d",
TimeUnit.MILLISECONDS.toDays(timeDiff),
TimeUnit.MILLISECONDS.toHours(timeDiff) % TimeUnit.HOURS.toHours(1),
TimeUnit.MILLISECONDS.toMinutes(timeDiff) % TimeUnit.HOURS.toMinutes(1),
TimeUnit.MILLISECONDS.toSeconds(timeDiff) % TimeUnit.MINUTES.toSeconds(1),
TimeUnit.MILLISECONDS.toMillis(timeDiff) % TimeUnit.SECONDS.toMillis(1));
} catch (ParseException e) {
e.printStackTrace();
}
System.out.println("State duration: " + timeDiffString);
}
return timeDiffString;
}
@Override
public void open(Configuration config) {
ValueStateDescriptor<ProductionState> descriptor = new ValueStateDescriptor<>(
"stateOfProduction",
TypeInformation.of(new TypeHint<ProductionState>() {}),
ProductionState.NOT_PRODUCING);
stateOfProduction = getRuntimeContext().getState(descriptor);
}
@Override
public void flatMap(ImaginePaperData ImaginePaperData, Collector<String> output) throws Exception {
List<String> warnings = new ArrayList<>();
JSONObject jObject = new JSONObject();
String productionTime = "0";
String nonProductionTime = "0";
// Data analysis
if (stateOfProduction == null || stateOfProduction.value() == ProductionState.NOT_PRODUCING && ImaginePaperData.actSpeedCl > 60.0) {
stateOfProduction.update(ProductionState.PRODUCING);
} else if (stateOfProduction.value() == ProductionState.PRODUCING && ImaginePaperData.actSpeedCl < 60.0) {
stateOfProduction.update(ProductionState.NOT_PRODUCING);
}
if(stateOfProduction.value() == ProductionState.PRODUCING) {
if (!nonProductionTimestamps.isEmpty()) {
System.out.println("Production has started again, non production timestamps cleared");
nonProductionTimestamps.clear();
}
productionTimestamps.add(ImaginePaperData.timestamp);
System.out.println(productionTimestamps);
productionTime = calcProductionTime(productionTimestamps);
} else {
if(!productionTimestamps.isEmpty()) {
System.out.println("Production has stopped, production timestamps cleared");
productionTimestamps.clear();
}
nonProductionTimestamps.add(ImaginePaperData.timestamp);
warnings.add("Production has stopped.");
System.out.println(nonProductionTimestamps);
//System.out.println("Production stopped");
nonProductionTime = calcProductionTime(nonProductionTimestamps);
}
// The rest is just JSON stuff
我可能必须在ListState中保存这两个时间戳列表吗？
编辑：因为另一位用户问，这是我得到的数据。
{'szenario': 'machine01', 'timestamp': '31.10.2018 09:18:39.432069', 'data': {1: 100.0, 2: 100.0, 101: 94.0, 102: 120.0, 103: 65.0}}
我期望的行为是，我的flink程序收集两个列表productionTimestamps和nonProductionTimestamps中的时间戳。然后，我希望我的calcProductionTime方法从第一个时间戳中减去列表中的最后一个时间戳，以获得从我第一次检测到机器"正在生产"/"不正在生产"到它停止"正在生产"/"不在生产"之间的持续时间。
我发现"看似随机"的时间戳的原因是ApacheFlink的并行执行。当并行度设置为>1时，事件的顺序不再得到保证。
我的快速解决方案是将程序的并行度设置为1，据我所知，这可以保证事件的顺序。