public String anomaly(String sensorinput) {
System.out.println("Kai: DL-UDF starting");
GenModel rawModel;
try {
rawModel = (hex.genmodel.GenModel) Class.forName(modelClassName).newInstance();
EasyPredictModelWrapper model = new EasyPredictModelWrapper(rawModel);
// Prepare input sensor data to be in correct data format for the autoencoder model (double[]):
String[] inputStringArray = sensorinput.split("#");
double[] doubleValues = Arrays.stream(inputStringArray)
.mapToDouble(Double::parseDouble)
.toArray();
RowData row = new RowData();
int j = 0;
for (String colName : rawModel.getNames()) {
row.put(colName, doubleValues[j]);
j++;
}
AutoEncoderModelPrediction p = model.predictAutoEncoder(row);
// System.out.println("original: " + java.util.Arrays.toString(p.original));
// System.out.println("reconstructedrowData: " + p.reconstructedRowData);
// System.out.println("reconstructed: " + java.util.Arrays.toString(p.reconstructed));
double sum = 0;
for (int i = 0; i < p.original.length; i++) {
sum += (p.original[i] - p.reconstructed[i]) * (p.original[i] - p.reconstructed[i]);
}
// Calculate Mean Square Error => High reconstruction error means anomaly
double mse = sum / p.original.length;
System.out.println("MSE: " + mse);
String mseString = "" + mse;
return (mseString);
} catch (InstantiationException | IllegalAccessException | ClassNotFoundException e) {
System.out.println(e.toString());
} catch (PredictException e) {
System.out.println(e.toString());
}
return null;
}
}
如何使用Apache Kafka和MQTT Proxy运行演示？
执行演示的所有步骤都在Github项目中描述。
你只需安装Confluent Platform，然后按照以下步骤部署UDF，创建MQTT事件并通过KSQL levera处理它们....
这里使用Mosquitto生成MQTT消息。 当然，也可以使用任何其他MQTT客户端。 这是开放和标准化协议的巨大好处。
閱讀更多 程序員小新人學習 的文章
