Fix field converter for none numerical values
@@ -18,12 +18,16 @@ package org.streampipes.processors.transformation.flink.processor.converter;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.util.Collector;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
import org.streampipes.vocabulary.XSD;
import java.util.Map;
public class FieldConverter implements FlatMapFunction<Map<String, Object>, Map<String, Object>> {
+ private static Logger LOG = LoggerFactory.getLogger(FieldConverter.class);
+
private String convertProperty;
private String targetDatatype;
@@ -34,8 +38,9 @@ public class FieldConverter implements FlatMapFunction<Map<String, Object>, Map<
@Override
- public void flatMap(Map<String, Object> in, Collector<Map<String, Object>> out) throws Exception {
+ public void flatMap(Map<String, Object> in, Collector<Map<String, Object>> out) {
String value = String.valueOf(in.get(convertProperty));
+ try {
if (targetDatatype.equals(XSD._float.toString())) {
in.put(convertProperty, Float.parseFloat(value));
} else {
@@ -43,5 +48,10 @@ public class FieldConverter implements FlatMapFunction<Map<String, Object>, Map<
}
out.collect(in);
+
+ } catch (NumberFormatException e) {
+ LOG.error("Field Converter could not convert value: " + value + " of event: " + in);
+ }
+
}
}
