add two UDFs: MasterTrain and MasterDetector (#9648)
diff --git a/library-udf/src/main/java/org/apache/iotdb/library/anomaly/UDTFMasterDetect.java b/library-udf/src/main/java/org/apache/iotdb/library/anomaly/UDTFMasterDetect.java new file mode 100644 index 0000000..af84561 --- /dev/null +++ b/library-udf/src/main/java/org/apache/iotdb/library/anomaly/UDTFMasterDetect.java
@@ -0,0 +1,98 @@ +package org.apache.iotdb.library.anomaly; + +import org.apache.iotdb.library.anomaly.util.MasterDetector; +import org.apache.iotdb.udf.api.UDTF; +import org.apache.iotdb.udf.api.access.Row; +import org.apache.iotdb.udf.api.collector.PointCollector; +import org.apache.iotdb.udf.api.customizer.config.UDTFConfigurations; +import org.apache.iotdb.udf.api.customizer.parameter.UDFParameterValidator; +import org.apache.iotdb.udf.api.customizer.parameter.UDFParameters; +import org.apache.iotdb.udf.api.customizer.strategy.RowByRowAccessStrategy; +import org.apache.iotdb.udf.api.type.Type; + +import java.util.ArrayList; +import java.util.Objects; + +public class UDTFMasterDetect implements UDTF { + + private MasterDetector masterDetector; + private int output_column; + private String output_type; + + @Override + public void validate(UDFParameterValidator validator) throws Exception { + for (int i = 0; i < validator.getParameters().getAttributes().size(); i++) { + validator.validateInputSeriesDataType(i, Type.DOUBLE, Type.FLOAT, Type.INT32, Type.INT64); + } + if (validator.getParameters().hasAttribute("k")) { + validator.validate( + k -> (int) k > 0, + "Parameter k should be a positive integer.", + validator.getParameters().getInt("k")); + } + if (validator.getParameters().hasAttribute("p")) { + validator.validate( + p -> (int) p > 0, + "Order p should be a positive integer.", + validator.getParameters().getInt("p")); + } + if (validator.getParameters().hasAttribute("output_column")) { + validator.validate( + output_column -> (int) output_column > 0, + "Parameter output_column should be a positive integer.", + validator.getParameters().getInt("output_column")); + } + if (validator.getParameters().hasAttribute("eta")) { + validator.validate( + eta -> (double) eta > 0, + "Parameter eta should be larger than 0.", + validator.getParameters().getDouble("eta")); + } + if (validator.getParameters().hasAttribute("beta")) { + validator.validate( + beta -> (double) beta > 0, + "Parameter beta should be larger than 0.", + validator.getParameters().getDouble("beta")); + } + } + + @Override + public void beforeStart(UDFParameters parameters, UDTFConfigurations configurations) + throws Exception { + configurations.setAccessStrategy(new RowByRowAccessStrategy()); + output_type = parameters.getStringOrDefault("output_type", "repair"); + if (output_type.equals("repairing")) configurations.setOutputDataType(Type.DOUBLE); + else configurations.setOutputDataType(Type.BOOLEAN); + int columnCnt = (parameters.getDataTypes().size() - 1) / 2; + int k = parameters.getIntOrDefault("k", -1); + int p = parameters.getIntOrDefault("p", -1); + double eta = parameters.getDoubleOrDefault("eta", 1.0); + double beta = parameters.getDoubleOrDefault("beta", 1.0); + output_column = parameters.getIntOrDefault("output_column", 1); + masterDetector = new MasterDetector(columnCnt, k, p, eta, beta); + } + + @Override + public void transform(Row row, PointCollector collector) throws Exception { + if (!masterDetector.isNullRow(row)) { + masterDetector.addRow(row); + } + } + + @Override + public void terminate(PointCollector collector) throws Exception { + masterDetector.detectAndRepair(); + ArrayList<ArrayList<Double>> td_repaired = masterDetector.getTd_repaired(); + ArrayList<Long> td_time = masterDetector.getTd_time(); + ArrayList<Boolean> anomalies_in_repaired = masterDetector.getAnomalies_in_repaired(); + if (Objects.equals(output_type, "repair")) { + for (int i = 0; i < td_repaired.size(); i++) { + collector.putDouble(td_time.get(i), td_repaired.get(i).get(output_column)); + } + } else { + for (int i = 0; i < anomalies_in_repaired.size(); i++) { + collector.putBoolean(td_time.get(i), anomalies_in_repaired.get(i)); + } + } + } +}
diff --git a/library-udf/src/main/java/org/apache/iotdb/library/anomaly/UDTFMasterTrain.java b/library-udf/src/main/java/org/apache/iotdb/library/anomaly/UDTFMasterTrain.java new file mode 100644 index 0000000..930326e --- /dev/null +++ b/library-udf/src/main/java/org/apache/iotdb/library/anomaly/UDTFMasterTrain.java
@@ -0,0 +1,67 @@ +package org.apache.iotdb.library.anomaly; + +import org.apache.iotdb.library.anomaly.util.MasterTrainUtil; +import org.apache.iotdb.udf.api.UDTF; +import org.apache.iotdb.udf.api.access.Row; +import org.apache.iotdb.udf.api.collector.PointCollector; +import org.apache.iotdb.udf.api.customizer.config.UDTFConfigurations; +import org.apache.iotdb.udf.api.customizer.parameter.UDFParameterValidator; +import org.apache.iotdb.udf.api.customizer.parameter.UDFParameters; +import org.apache.iotdb.udf.api.customizer.strategy.RowByRowAccessStrategy; +import org.apache.iotdb.udf.api.type.Type; + +import java.util.ArrayList; + +public class UDTFMasterTrain implements UDTF { + + private MasterTrainUtil masterTrainUtil; + + private int columnCnt; + + @Override + public void validate(UDFParameterValidator validator) throws Exception { + for (int i = 0; i < validator.getParameters().getAttributes().size(); i++) { + validator.validateInputSeriesDataType(i, Type.DOUBLE, Type.FLOAT, Type.INT32, Type.INT64); + } + if (validator.getParameters().hasAttribute("p")) { + validator.validate( + p -> (int) p > 0, + "Order p should be a positive integer.", + validator.getParameters().getInt("p")); + } + if (validator.getParameters().hasAttribute("eta")) { + validator.validate( + eta -> (double) eta > 0, + "Parameter eta should be larger than 0.", + validator.getParameters().getDouble("eta")); + } + } + + @Override + public void beforeStart(UDFParameters parameters, UDTFConfigurations configurations) + throws Exception { + configurations.setAccessStrategy(new RowByRowAccessStrategy()); + configurations.setOutputDataType(Type.DOUBLE); + columnCnt = parameters.getDataTypes().size() / 2; + int p = parameters.getIntOrDefault("p", -1); + double eta = parameters.getDoubleOrDefault("eta", 1.0); + masterTrainUtil = new MasterTrainUtil(columnCnt, p, eta); + } + + @Override + public void transform(Row row, PointCollector collector) throws Exception { + if (!masterTrainUtil.isNullRow(row)) { + masterTrainUtil.addRow(row); + } + } + + @Override + public void terminate(PointCollector collector) throws Exception { + masterTrainUtil.train(); + ArrayList<Double> coeffs_one_column = masterTrainUtil.coeffsInOneColumn(); + ArrayList<Long> td_time = masterTrainUtil.getTd_time(); + for (int i = 0; i < coeffs_one_column.size(); i++) { + collector.putDouble(td_time.get(i), coeffs_one_column.get(i)); + } + } +}
diff --git a/library-udf/src/main/java/org/apache/iotdb/library/anomaly/util/KDTreeUtil.java b/library-udf/src/main/java/org/apache/iotdb/library/anomaly/util/KDTreeUtil.java new file mode 100644 index 0000000..c44de23 --- /dev/null +++ b/library-udf/src/main/java/org/apache/iotdb/library/anomaly/util/KDTreeUtil.java
@@ -0,0 +1,116 @@ +package org.apache.iotdb.library.anomaly.util; + +import java.util.ArrayList; +import java.util.Comparator; + +public class KDTreeUtil { + private Node root; + + private static class Node { + ArrayList<Double> data; + int depth; + int index; + Node left; + Node right; + + Node(ArrayList<Double> data, int depth, int index) { + this.data = data; + this.depth = depth; + this.index = index; + } + } + + public void buildTree(ArrayList<ArrayList<Double>> dataset) { + if (dataset == null || dataset.isEmpty()) { + throw new IllegalArgumentException("Dataset is null or empty"); + } + root = buildTree(dataset, 0, dataset.size() - 1, 0); + } + + private Node buildTree(ArrayList<ArrayList<Double>> dataset, int left, int right, int depth) { + if (left > right) { + return null; + } + int mid = (left + right) / 2; + ArrayList<Double> data = dataset.get(mid); + Node node = new Node(data, depth, mid); + int nextDepth = (depth + 1) % data.size(); + node.left = buildTree(dataset, left, mid - 1, nextDepth); + node.right = buildTree(dataset, mid + 1, right, nextDepth); + return node; + } + + public ArrayList<ArrayList<Double>> findKNearestNeighbors(ArrayList<Double> query, int k) { + if (query == null || query.isEmpty()) { + throw new IllegalArgumentException("Query is null or empty"); + } + if (k <= 0) { + throw new IllegalArgumentException("k must be a positive integer"); + } + ArrayList<ArrayList<Double>> knnList = new ArrayList<>(); + findKNN(root, query, k, knnList); + return knnList; + } + + private void findKNN( + Node node, ArrayList<Double> query, int k, ArrayList<ArrayList<Double>> knnList) { + if (node == null) { + return; + } + double dist = euclideanDistance(node.data, query); + if (knnList.size() < k || dist < euclideanDistance(knnList.get(knnList.size() - 1), query)) { + knnList.add(node.data); + knnList.sort(Comparator.comparingDouble(p -> euclideanDistance(p, query))); + if (knnList.size() > k) { + knnList.remove(k); + } + } + double axisDistance = + node.data.get(node.depth % node.data.size()) - query.get(node.depth % node.data.size()); + Node nearerNode = (axisDistance <= 0) ? node.left : node.right; + Node fartherNode = (axisDistance <= 0) ? node.right : node.left; + findKNN(nearerNode, query, k, knnList); + if (knnList.size() < k + || Math.abs(axisDistance) < euclideanDistance(knnList.get(knnList.size() - 1), query)) { + findKNN(fartherNode, query, k, knnList); + } + } + + private static double euclideanDistance(ArrayList<Double> p1, ArrayList<Double> p2) { + if (p1 == null || p2 == null || p1.size() != p2.size()) { + throw new IllegalArgumentException("Points are null or have different dimensions"); + } + double sum = 0.0; + for (int i = 0; i < p1.size(); i++) { + double diff = p1.get(i) - p2.get(i); + sum += diff * diff; + } + return Math.sqrt(sum); + } + + public ArrayList<Double> findTheNearestNeighbor(ArrayList<Double> tuple) { + if (tuple == null || tuple.isEmpty()) { + throw new IllegalArgumentException("Tuple is null or empty"); + } + return findNN(root, tuple, root.data); + } + + private ArrayList<Double> findNN(Node node, ArrayList<Double> query, ArrayList<Double> nn) { + if (node == null) { + return nn; + } + double dist = euclideanDistance(node.data, query); + if (dist < euclideanDistance(nn, query)) { + nn = node.data; + } + double axisDistance = + node.data.get(node.depth % node.data.size()) - query.get(node.depth % node.data.size()); + Node nearerNode = (axisDistance <= 0) ? node.left : node.right; + Node fartherNode = (axisDistance <= 0) ? node.right : node.left; + nn = findNN(nearerNode, query, nn); + if (Math.abs(axisDistance) < euclideanDistance(nn, query)) { + nn = findNN(fartherNode, query, nn); + } + return nn; + } +}
diff --git a/library-udf/src/main/java/org/apache/iotdb/library/anomaly/util/MasterDetector.java b/library-udf/src/main/java/org/apache/iotdb/library/anomaly/util/MasterDetector.java new file mode 100644 index 0000000..31c20fd --- /dev/null +++ b/library-udf/src/main/java/org/apache/iotdb/library/anomaly/util/MasterDetector.java
@@ -0,0 +1,361 @@ +package org.apache.iotdb.library.anomaly.util; + +import org.apache.iotdb.library.util.Util; +import org.apache.iotdb.udf.api.access.Row; + +import java.math.BigDecimal; +import java.util.ArrayList; + +public class MasterDetector { + private final ArrayList<ArrayList<Double>> td = new ArrayList<>(); + private final ArrayList<ArrayList<Double>> td_repaired = new ArrayList<>(); + private final ArrayList<Boolean> td_anomalies = new ArrayList<>(); + private final ArrayList<Boolean> anomalies_in_repaired = new ArrayList<>(); + private final ArrayList<ArrayList<Double>> md = new ArrayList<>(); + private final ArrayList<Double> coeffs_one_column = new ArrayList<>(); + private final ArrayList<Long> td_time = new ArrayList<>(); + private int[] initial_window; + private final int columnCnt; + private int k; + private int p; + private double[] std; + private KDTreeUtil kdTreeUtil; + + private VAR prediction_model; + + private double eta; + private double beta; + + public MasterDetector(int columnCnt, int k, int p, double eta, double beta) { + this.columnCnt = columnCnt; + this.k = k; + this.p = p; + this.eta = eta; + this.beta = beta; + } + + public boolean isNullRow(Row row) { + boolean flag = true; + for (int i = 0; i < row.size(); i++) { + if (!row.isNull(i)) { + flag = false; + break; + } + } + return flag; + } + + public void addRow(Row row) throws Exception { + ArrayList<Double> tt = new ArrayList<>(); // time-series tuple + boolean containsNotNull = false; + for (int i = 0; i < this.columnCnt; i++) { + if (!row.isNull(i)) { + containsNotNull = true; + BigDecimal bd = BigDecimal.valueOf(Util.getValueAsDouble(row, i)); + tt.add(bd.doubleValue()); + } else { + tt.add(null); + } + } + if (containsNotNull) { + td.add(tt); + td_time.add(row.getTime()); + } + + ArrayList<Double> mt = new ArrayList<>(); // master tuple + containsNotNull = false; + for (int i = this.columnCnt; i < this.columnCnt * 2; i++) { + if (!row.isNull(i)) { + containsNotNull = true; + BigDecimal bd = BigDecimal.valueOf(Util.getValueAsDouble(row, i)); + mt.add(bd.doubleValue()); + } else { + mt.add(null); + } + } + if (containsNotNull) { + md.add(mt); + } + + int i = this.columnCnt * 2; + if (!row.isNull(i)) { + coeffs_one_column.add(Util.getValueAsDouble(row, i)); + } + } + + public void buildKDTree() { + this.kdTreeUtil = new KDTreeUtil(); + this.kdTreeUtil.buildTree(this.md); + } + + public double delta(ArrayList<Double> t_tuple, ArrayList<Double> m_tuple) { + double distance = 0d; + for (int pos = 0; pos < columnCnt; pos++) { + double temp = t_tuple.get(pos) - m_tuple.get(pos); + temp = temp / std[pos]; + distance += temp * temp; + } + distance = Math.sqrt(distance); + return distance; + } + + public void fillNullValue() { + for (int i = 0; i < columnCnt; i++) { + double temp = this.td.get(0).get(i); + for (ArrayList<Double> arrayList : this.td) { + if (arrayList.get(i) == null) { + arrayList.set(i, temp); + } else { + temp = arrayList.get(i); + } + } + } + } + + private double varianceImperative(double[] value) { + double average = 0.0; + int cnt = 0; + for (double p : value) { + if (!Double.isNaN(p)) { + cnt += 1; + average += p; + } + } + if (cnt == 0) { + return 0d; + } + average /= cnt; + + double variance = 0.0; + for (double p : value) { + if (!Double.isNaN(p)) { + variance += (p - average) * (p - average); + } + } + return variance / cnt; + } + + private double[] getColumn(int pos) { + double[] column = new double[this.td.size()]; + for (int i = 0; i < this.td.size(); i++) { + column[i] = this.td.get(i).get(pos); + } + return column; + } + + public void call_std() { + this.std = new double[this.columnCnt]; + for (int i = 0; i < this.columnCnt; i++) { + std[i] = Math.sqrt(varianceImperative(getColumn(i))); + } + } + + public boolean checkConsistency(ArrayList<Double> tuple) { + ArrayList<Double> NN = kdTreeUtil.findTheNearestNeighbor(tuple); + double delta = delta(tuple, NN); + if (delta > eta) { + return false; + } else return true; + } + + public void getOriginalAnomaliesAndLearnModel() { + for (int i = 0; i < this.td.size(); i++) { + ArrayList<Double> tuple = this.td.get(i); + boolean isNormal = checkConsistency(tuple); + td_anomalies.add(!isNormal); + } + this.prediction_model = new VAR(columnCnt); + this.prediction_model.fitCoeffs(coeffs_one_column, columnCnt); + } + + public void findInitialWindow(int p) { + initial_window = new int[2]; + int left = 0; + int right = 0; + for (int i = 0; i < td_anomalies.size(); i++) { + if (right - left + 1 == p) { + initial_window[0] = left; + initial_window[1] = right; + break; + } + if (td_anomalies.get(i) == Boolean.TRUE) { + left = i + 1; + right = i + 1; + } else { + right++; + } + } + } + + public ArrayList<ArrayList<Double>> getWindow(ArrayList<ArrayList<Double>> data, int i, int p) { + if (i < p) { + System.out.println("ERROR: i must be greater than p."); + return new ArrayList<>(); + } + ArrayList<ArrayList<Double>> W = new ArrayList<>(); + for (int j = i - p; j < i; j++) { + W.add(data.get(j)); + } + return W; + } + + public double calForwardPredictionLoss(int i, int p, ArrayList<Double> candidate_for_i) { + double sum_prediction_loss = 0.0; + for (int index = 0; index < p; index++) { + if (index == 0) { + ArrayList<ArrayList<Double>> W_repaired = getWindow(this.td_repaired, i, p); + ArrayList<Double> prediction_for_i = prediction_model.predict(W_repaired); + sum_prediction_loss += delta(prediction_for_i, candidate_for_i); + } else { + if (td_anomalies.get(i + index) == Boolean.TRUE) { + break; + } + + ArrayList<ArrayList<Double>> W_repaired = getWindow(this.td_repaired, i, p - index); + W_repaired.add(candidate_for_i); + for (int j = 1; j < index; j++) { + W_repaired.add(td.get(i + j)); + } + ArrayList<Double> prediction_for_i = prediction_model.predict(W_repaired); + sum_prediction_loss += delta(prediction_for_i, td.get(i + index)); + } + } + return sum_prediction_loss; + } + + public double calBackwardPredictionLoss(int i, int p, ArrayList<Double> candidate_for_i) { + double sum_prediction_loss = 0.0; + for (int index = 0; index < p; index++) { + if (index == 0) { + ArrayList<ArrayList<Double>> W_repaired = getWindow(this.td_repaired, i + p, p); + W_repaired.set(0, candidate_for_i); + ArrayList<Double> prediction_for_i = prediction_model.predict(W_repaired); + sum_prediction_loss += delta(prediction_for_i, td_repaired.get(i + p)); + } else { + if (i - index < 0 || td_anomalies.get(i - index) == Boolean.TRUE) { + break; + } + ArrayList<ArrayList<Double>> W_repaired = getWindow(this.td_repaired, i + p - index, p); + W_repaired.set(index, candidate_for_i); + ArrayList<Double> prediction_for_i = prediction_model.predict(W_repaired); + sum_prediction_loss += delta(prediction_for_i, td_repaired.get(i + p)); + } + } + return sum_prediction_loss; + } + + public void forwardRepairing(int p) { + int i = initial_window[1] + 1; + + while (i < td.size()) { + ArrayList<ArrayList<Double>> W_repaired = getWindow(this.td_repaired, i, p); + ArrayList<Double> x_repaired_predicted = prediction_model.predict(W_repaired); + ArrayList<Double> optimal_repair = new ArrayList<>(); + if (td_anomalies.get(i) == Boolean.TRUE) { + ArrayList<ArrayList<Double>> candidates = + this.kdTreeUtil.findKNearestNeighbors(x_repaired_predicted, this.k); + // find the optimal repair + double min_dis = Double.MAX_VALUE; + for (ArrayList<Double> candidate : candidates) { + double prediction_loss = calForwardPredictionLoss(i, p, candidate); + if (prediction_loss < min_dis) { + min_dis = prediction_loss; + optimal_repair = candidate; + } + } + this.td_repaired.add(optimal_repair); + } else { + optimal_repair = td.get(i); + this.td_repaired.add(optimal_repair); + } + if (delta(x_repaired_predicted, optimal_repair) > beta) { + this.anomalies_in_repaired.add(Boolean.TRUE); + } else { + this.anomalies_in_repaired.add(Boolean.FALSE); + } + i++; + } + } + + public void backwardRepairing(int p) { + int i = initial_window[0] - 1; + if (i < 0) { + return; + } + + while (i >= 0) { + ArrayList<Double> optimal_repair = new ArrayList<>(); + if (td_anomalies.get(i) == Boolean.TRUE) { + ArrayList<ArrayList<Double>> candidates = + this.kdTreeUtil.findKNearestNeighbors(this.td_repaired.get(i + 1), k); + double min_dis = Double.MAX_VALUE; + for (ArrayList<Double> candidate : candidates) { + double prediction_loss = calBackwardPredictionLoss(i, p, candidate); + if (prediction_loss < min_dis) { + min_dis = prediction_loss; + optimal_repair = candidate; + } + } + this.td_repaired.set(i, optimal_repair); + } else { + optimal_repair = td.get(i); + this.td_repaired.set(i, optimal_repair); + } + + if (delta(optimal_repair, td_repaired.get(i + p)) > beta) { + this.anomalies_in_repaired.set(i, Boolean.TRUE); + } else { + this.anomalies_in_repaired.set(i, Boolean.FALSE); + } + i--; + } + } + + public void detectAndRepair() { + fillNullValue(); + buildKDTree(); + call_std(); + ArrayList<Double> zero_tuple = new ArrayList<>(); + zero_tuple.add(0.0); + zero_tuple.add(0.0); + zero_tuple.add(0.0); + for (int i = 0; i < 10; i++) { + this.td.set(i, zero_tuple); + } + getOriginalAnomaliesAndLearnModel(); + findInitialWindow(p); + System.out.println(initial_window[0] + " " + initial_window[1]); + + for (int j = 0; j <= initial_window[1]; j++) { + this.td_repaired.add(this.td.get(j)); + this.anomalies_in_repaired.add(Boolean.FALSE); + } + + backwardRepairing(p); + forwardRepairing(p); + } + + public ArrayList<ArrayList<Double>> getMd() { + return md; + } + + public ArrayList<ArrayList<Double>> getTd() { + return td; + } + + public ArrayList<Long> getTd_time() { + return td_time; + } + + public ArrayList<Double> getCoeffs_one_column() { + return coeffs_one_column; + } + + public ArrayList<ArrayList<Double>> getTd_repaired() { + return td_repaired; + } + + public ArrayList<Boolean> getAnomalies_in_repaired() { + return anomalies_in_repaired; + } +}
diff --git a/library-udf/src/main/java/org/apache/iotdb/library/anomaly/util/MasterTrainUtil.java b/library-udf/src/main/java/org/apache/iotdb/library/anomaly/util/MasterTrainUtil.java new file mode 100644 index 0000000..1957e2f --- /dev/null +++ b/library-udf/src/main/java/org/apache/iotdb/library/anomaly/util/MasterTrainUtil.java
@@ -0,0 +1,200 @@ +package org.apache.iotdb.library.anomaly.util; + +import org.apache.iotdb.library.util.Util; +import org.apache.iotdb.udf.api.access.Row; + +import java.math.BigDecimal; +import java.util.ArrayList; + +public class MasterTrainUtil { + private final ArrayList<ArrayList<Double>> td = new ArrayList<>(); + private final ArrayList<ArrayList<Double>> md = new ArrayList<>(); + private final ArrayList<Long> td_time = new ArrayList<>(); + private final int columnCnt; + private int p; + private double[] std; + private KDTreeUtil kdTreeUtil; + + private VAR prediction_model; + + private double eta; + + public MasterTrainUtil(int columnCnt, int p, double eta) { + this.columnCnt = columnCnt; + this.p = p; + this.eta = eta; + } + + public boolean isNullRow(Row row) { + boolean flag = true; + for (int i = 0; i < row.size(); i++) { + if (!row.isNull(i)) { + flag = false; + break; + } + } + return flag; + } + + public void addRow(Row row) throws Exception { + ArrayList<Double> tt = new ArrayList<>(); // time-series tuple + boolean containsNotNull = false; + for (int i = 0; i < this.columnCnt; i++) { + if (!row.isNull(i)) { + containsNotNull = true; + BigDecimal bd = BigDecimal.valueOf(Util.getValueAsDouble(row, i)); + tt.add(bd.doubleValue()); + } else { + tt.add(null); + } + } + if (containsNotNull) { + td.add(tt); + td_time.add(row.getTime()); + } + + ArrayList<Double> mt = new ArrayList<>(); // master tuple + containsNotNull = false; + for (int i = this.columnCnt; i < row.size(); i++) { + if (!row.isNull(i)) { + containsNotNull = true; + BigDecimal bd = BigDecimal.valueOf(Util.getValueAsDouble(row, i)); + mt.add(bd.doubleValue()); + } else { + mt.add(null); + } + } + if (containsNotNull) { + md.add(mt); + } + } + + public void buildKDTree() { + this.kdTreeUtil = new KDTreeUtil(); + this.kdTreeUtil.buildTree(this.md); + } + + public double delta(ArrayList<Double> t_tuple, ArrayList<Double> m_tuple) { + double distance = 0d; + for (int pos = 0; pos < columnCnt; pos++) { + double temp = t_tuple.get(pos) - m_tuple.get(pos); + temp = temp / std[pos]; + distance += temp * temp; + } + distance = Math.sqrt(distance); + return distance; + } + + public void fillNullValue() { + for (int i = 0; i < columnCnt; i++) { + double temp = this.td.get(0).get(i); + for (ArrayList<Double> arrayList : this.td) { + if (arrayList.get(i) == null) { + arrayList.set(i, temp); + } else { + temp = arrayList.get(i); + } + } + } + } + + private double varianceImperative(double[] value) { + double average = 0.0; + int cnt = 0; + for (double p : value) { + if (!Double.isNaN(p)) { + cnt += 1; + average += p; + } + } + if (cnt == 0) { + return 0d; + } + average /= cnt; + + double variance = 0.0; + for (double p : value) { + if (!Double.isNaN(p)) { + variance += (p - average) * (p - average); + } + } + return variance / cnt; + } + + private double[] getColumn(int pos) { + double[] column = new double[this.td.size()]; + for (int i = 0; i < this.td.size(); i++) { + column[i] = this.td.get(i).get(pos); + } + return column; + } + + public void call_std() { + this.std = new double[this.columnCnt]; + for (int i = 0; i < this.columnCnt; i++) { + std[i] = Math.sqrt(varianceImperative(getColumn(i))); + } + } + + public boolean checkConsistency(ArrayList<Double> tuple) { + ArrayList<Double> NN = kdTreeUtil.findTheNearestNeighbor(tuple); + double delta = delta(tuple, NN); + if (delta > eta) { + return false; + } else return true; + } + + public void getOriginalAnomaliesAndTrainModel() { + int left = 0; + int right = 0; + ArrayList<ArrayList<Double>> learning_samples = new ArrayList<>(); + for (int i = 0; i < this.td.size(); i++) { + ArrayList<Double> tuple = this.td.get(i); + boolean isNormal = checkConsistency(tuple); + if (right - left + 1 == p) { + for (int j = left; j <= right; j++) { + learning_samples.add(this.td.get(j)); + } + left = i + 1; + right = i + 1; + continue; + } + if (isNormal) { + right++; + } else { + left = i + 1; + right = i + 1; + } + } + this.prediction_model = new VAR(columnCnt); + this.prediction_model.fit(learning_samples); + } + + public void train() { + fillNullValue(); + buildKDTree(); + call_std(); + getOriginalAnomaliesAndTrainModel(); + } + + public ArrayList<Double> coeffsInOneColumn() { + ArrayList<ArrayList<Double>> coeffs = this.prediction_model.getCoeffs(); + ArrayList<Double> coeffs_one_column = new ArrayList<>(); + for (ArrayList<Double> coeff : coeffs) { + coeffs_one_column.addAll(coeff); + } + return coeffs_one_column; + } + + public ArrayList<ArrayList<Double>> getMd() { + return md; + } + + public ArrayList<ArrayList<Double>> getTd() { + return td; + } + + public ArrayList<Long> getTd_time() { + return td_time; + } +}
diff --git a/library-udf/src/main/java/org/apache/iotdb/library/anomaly/util/VAR.java b/library-udf/src/main/java/org/apache/iotdb/library/anomaly/util/VAR.java new file mode 100644 index 0000000..13d3cd0 --- /dev/null +++ b/library-udf/src/main/java/org/apache/iotdb/library/anomaly/util/VAR.java
@@ -0,0 +1,217 @@ +package org.apache.iotdb.library.anomaly.util; + +import java.util.ArrayList; + +public class VAR { + private final int p; + private ArrayList<ArrayList<Double>> coeffs; + + public VAR(int p) { + this.p = p; + this.coeffs = new ArrayList<>(); + } + + // Train the model with data to get coefficients + public void fit(ArrayList<ArrayList<Double>> data) { + int n = data.size(); + int k = data.get(0).size(); + + ArrayList<ArrayList<Double>> X = new ArrayList<>(); + + // Construct the data matrix X + for (int i = p; i < n; i++) { + ArrayList<Double> x = new ArrayList<>(); + for (int j = 0; j < p; j++) { + x.addAll(data.get(i - j - 1)); + } + X.add(x); + } + + // Compute the coefficients using OLS + Matrix Xmat = new Matrix(X); + ArrayList<ArrayList<Double>> Yarry = new ArrayList<>(); + for (int i = p; i < n; i++) { + Yarry.add(data.get(i)); + } + Matrix Ymat = new Matrix(Yarry); + Matrix XtX = Xmat.transpose().multiply(Xmat); + Matrix XtY = Xmat.transpose().multiply(Ymat); + Matrix beta = XtX.solve(XtY); + this.coeffs = beta.transpose().getData(); + } + + public void fitCoeffs(ArrayList<Double> coeffs_in_one_column, int column_cnt) { + ArrayList<ArrayList<Double>> coeffs = new ArrayList<>(); + int single_size = coeffs_in_one_column.size() / column_cnt; + for (int i = 0; i < coeffs_in_one_column.size(); i += single_size) { + ArrayList<Double> temp = new ArrayList<>(); + for (int j = 0; j < single_size; j++) { + temp.add(coeffs_in_one_column.get(i + j)); + } + coeffs.add(temp); + } + this.coeffs = coeffs; + } + + // One step of prediction based on window. Window has p tuples. + // Return the prediction result. + public ArrayList<Double> predict(ArrayList<ArrayList<Double>> window) { + int k = window.get(0).size(); + ArrayList<Double> x = new ArrayList<>(); + for (int i = 0; i < p; i++) { + x.addAll(window.get(p - i - 1)); + } + double[] yhat = new double[k]; + ArrayList<Double> prediction_tuple = new ArrayList<>(); + for (int j = 0; j < k; j++) { + for (int i = 0; i < x.size(); i++) { + yhat[j] += x.get(i) * coeffs.get(j).get(i); + } + prediction_tuple.add(yhat[j]); + } + + return prediction_tuple; + } + + public ArrayList<ArrayList<Double>> getCoeffs() { + return coeffs; + } + + // Helper class for matrix operations + public class Matrix { + public final int m; + public final int n; + public final ArrayList<ArrayList<Double>> data; + + public Matrix(int m, int n) { + this.m = m; + this.n = n; + this.data = new ArrayList<ArrayList<Double>>(m); + for (int i = 0; i < m; i++) { + ArrayList<Double> row = new ArrayList<Double>(n); + for (int j = 0; j < n; j++) { + row.add(0.0); + } + this.data.add(row); + } + } + + public Matrix(ArrayList<ArrayList<Double>> data) { + this.m = data.size(); + this.n = data.get(0).size(); + this.data = new ArrayList<ArrayList<Double>>(m); + for (int i = 0; i < m; i++) { + ArrayList<Double> row = new ArrayList<Double>(n); + for (int j = 0; j < n; j++) { + row.add(data.get(i).get(j)); + } + this.data.add(row); + } + } + + public Matrix transpose() { + Matrix A = this; + Matrix C = new Matrix(n, m); + for (int i = 0; i < m; i++) { + for (int j = 0; j < n; j++) { + C.data.get(j).set(i, A.data.get(i).get(j)); + } + } + return C; + } + + public Matrix multiply(Matrix B) { + Matrix A = this; + if (A.n != B.m) { + throw new IllegalArgumentException("Matrix dimensions don't match"); + } + Matrix C = new Matrix(A.m, B.n); + for (int i = 0; i < C.m; i++) { + for (int j = 0; j < C.n; j++) { + for (int k = 0; k < A.n; k++) { + C.data + .get(i) + .set(j, C.data.get(i).get(j) + A.data.get(i).get(k) * B.data.get(k).get(j)); + } + } + } + return C; + } + + public ArrayList<ArrayList<Double>> getArray() { + return data; + } + + public Matrix solve(Matrix B) { + Matrix A = this; + if (A.m != A.n || A.m != B.m) { + throw new IllegalArgumentException("Matrix dimensions don't match"); + } + + int n = A.n; + Matrix[] LU = A.lu(); + Matrix L = LU[0]; + Matrix U = LU[1]; + + // Solve LY = B using forward substitution + Matrix Y = new Matrix(n, B.n); + for (int j = 0; j < B.n; j++) { + for (int i = 0; i < n; i++) { + Y.data.get(i).set(j, B.data.get(i).get(j)); + for (int k = 0; k < i; k++) { + Y.data + .get(i) + .set(j, Y.data.get(i).get(j) - L.data.get(i).get(k) * Y.data.get(k).get(j)); + } + } + } + + // Solve UX = Y using backward substitution + Matrix X = new Matrix(n, B.n); + for (int j = 0; j < B.n; j++) { + for (int i = n - 1; i >= 0; i--) { + X.data.get(i).set(j, Y.data.get(i).get(j)); + for (int k = i + 1; k < n; k++) { + X.data + .get(i) + .set(j, X.data.get(i).get(j) - U.data.get(i).get(k) * X.data.get(k).get(j)); + } + X.data.get(i).set(j, X.data.get(i).get(j) / U.data.get(i).get(i)); + } + } + return X; + } + + public Matrix[] lu() { + Matrix A = this; + if (A.m != A.n) { + throw new IllegalArgumentException("Matrix dimensions don't match"); + } + + Matrix L = new Matrix(A.m, A.n); + Matrix U = new Matrix(A.m, A.n); + for (int j = 0; j < A.n; j++) { + L.data.get(j).set(j, 1.0); + for (int i = 0; i < j + 1; i++) { + double s1 = 0.0; + for (int k = 0; k < i; k++) { + s1 += U.data.get(k).get(j) * L.data.get(i).get(k); + } + U.data.get(i).set(j, A.data.get(i).get(j) - s1); + } + for (int i = j + 1; i < A.n; i++) { + double s2 = 0.0; + for (int k = 0; k < j; k++) { + s2 += U.data.get(k).get(j) * L.data.get(i).get(k); + } + L.data.get(i).set(j, (A.data.get(i).get(j) - s2) / U.data.get(j).get(j)); + } + } + return new Matrix[] {L, U}; + } + + public ArrayList<ArrayList<Double>> getData() { + return data; + } + } +}